code: @Override
    public byte[] call(final byte[] memento) throws Exception {
      final ExecutorService schedulerThread = Executors.newSingleThreadExecutor();
      final ExecutorService commandExecutor = Executors.newFixedThreadPool(numOfThreads);
      final ConcurrentMap<Integer, Future> futures = new ConcurrentHashMap<>();
  
      // Scheduling thread starts
      schedulerThread.execute(new Runnable() {
        @SuppressWarnings("InfiniteLoopStatement") // Scheduler is supposed to run forever.
        @Override
        public void run() {
          while (true) {
            // Scheduler Thread: Pick a command to execute (For now, simple FIFO order)
            final byte[] message;
            try {
              message = pendingRequests.takeFirst();
            } catch (InterruptedException e) {
              throw new RuntimeException(e);
            }
  
            // Command Executor: Deserialize the command
            final MasterToWorkerRequest masterToWorkerRequest = (MasterToWorkerRequest)kryoUtils.deserialize(message);
  
            switch (masterToWorkerRequest.getType()) {
              case AggregateTasklets:
                final TaskletAggregationRequest taskletAggregationRequest =
                    (TaskletAggregationRequest) masterToWorkerRequest;
                aggregates.put(taskletAggregationRequest.getAggregateFunctionId(),
                    new AggregateContainer(heartBeatTriggerManager, kryoUtils, workerReports,
                        taskletAggregationRequest));
                break;
              case ExecuteAggregateTasklet:
                executeAggregateTasklet(commandExecutor, masterToWorkerRequest);
                break;
              case ExecuteTasklet:
                executeTasklet(commandExecutor, futures, masterToWorkerRequest);
                break;
              case CancelTasklet:
                final TaskletCancellationRequest cancellationRequest = (TaskletCancellationRequest) masterToWorkerRequest;
                LOG.log(Level.FINE, "Cancelling Tasklet with ID {0}.", cancellationRequest.getTaskletId());
                final Future future = futures.get(cancellationRequest.getTaskletId());
                if (future != null) {
                  future.cancel(true);
                }
                break;
              default:
                throw new RuntimeException("Unknown Command");
            }
          }
        }
      });
  
      terminated.await();
      return null;
    }
docs: Starts the scheduler and executor and waits until termination.