code: public void scheduleTasklet(final int taskletId) {
      synchronized (stateLock) {
        // If there are tasklets are pending to be executed, then that means that a
        // timer has already been scheduled for an aggregation.
        if (!outstandingTasklets()) {
          timer.schedule(new Runnable() {
            @Override
            public void run() {
              aggregateTasklets(AggregateTriggerType.ALARM);
              synchronized (stateLock) {
                // On the callback, if there are tasklets pending to be executed, that means that this alarm
                // was triggered by a previous alarm, so we should continue to trigger more alarms. Otherwise
                // we are done with tasklets for this aggregation function for now.
                // If more tasklets for this aggregation function arrive, it will be triggered by the outer
                // call to timer.schedule.
                if (outstandingTasklets()) {
                  timer.schedule(
                      this, taskletAggregationRequest.getPolicy().getPeriodMilliseconds(), TimeUnit.MILLISECONDS);
                }
              }
            }
          }, taskletAggregationRequest.getPolicy().getPeriodMilliseconds(), TimeUnit.MILLISECONDS);
        }
  
        // Add to pending tasklets, such that on the callback the timer can be refreshed.
        if (!pendingTasklets.containsKey(taskletId)) {
          pendingTasklets.put(taskletId, 0);
        }
  
        pendingTasklets.put(taskletId, pendingTasklets.get(taskletId) + 1);
      }
    }
docs: Schedule aggregation tasks on a Timer. Creates a new timer schedule for triggering the aggregation function
  if this is the first time the aggregation function has tasklets scheduled on it.
  Adds the Tasklet to pending Tasklets.