code: @Override
      public Entry<BatchPartitionWorkUnit, Future<?>> startPartition(PartitionPlanConfig partitionPlanConfig,
                                                                     Step step,
                                                                     PartitionReplyQueue partitionReplyQueue,
                                                                     boolean isRemoteDispatch) {
  
          BatchPartitionWorkUnit workUnit = createPartitionWorkUnit(partitionPlanConfig, step, partitionReplyQueue, isRemoteDispatch);
  
          Future<?> futureWork = runPartition(workUnit);
  
          // TODO: issue "partition.started" message ?
  
          return new AbstractMap.SimpleEntry<BatchPartitionWorkUnit, Future<?>>(workUnit, futureWork);
      }
docs: Create the BatchPartitionWorkUnit and start the sub-job partition thread.
  
  @return A Map Entry with BatchPartitionWorkUnit as key and a Future WorkUnit as value