@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;
  }
@ 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 < > ( ) ; schedulerThread . execute ( new Runnable ( ) { @ SuppressWarnings ( "InfiniteLoopStatement" ) @ Override public void run ( ) { while ( true ) { final byte [ ] message ; try { message = pendingRequests . takeFirst ( ) ; } catch ( InterruptedException e ) { throw new RuntimeException ( e ) ; } 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 ; }
Starts the scheduler and executor and waits until termination.