@Override
  public DataflowMapTaskExecutor create(
      MutableNetwork<Node, Edge> network,
      PipelineOptions options,
      String stageName,
      ReaderFactory readerFactory,
      SinkFactory sinkFactory,
      DataflowExecutionContext<?> executionContext,
      CounterSet counterSet,
      IdGenerator idGenerator) {

    // Swap out all the InstructionOutput nodes with OutputReceiver nodes
    Networks.replaceDirectedNetworkNodes(
        network, createOutputReceiversTransform(stageName, counterSet));

    // Swap out all the ParallelInstruction nodes with Operation nodes
    Networks.replaceDirectedNetworkNodes(
        network,
        createOperationTransformForParallelInstructionNodes(
            stageName, network, options, readerFactory, sinkFactory, executionContext));

    // Collect all the operations within the network and attach all the operations as receivers
    // to preceding output receivers.
    List<Operation> topoSortedOperations = new ArrayList<>();
    for (OperationNode node :
        Iterables.filter(Networks.topologicalOrder(network), OperationNode.class)) {
      topoSortedOperations.add(node.getOperation());

      for (Node predecessor :
          Iterables.filter(network.predecessors(node), OutputReceiverNode.class)) {
        ((OutputReceiverNode) predecessor)
            .getOutputReceiver()
            .addOutput((Receiver) node.getOperation());
      }
    }

    if (LOG.isDebugEnabled()) {
      LOG.info("Map task network: {}", Networks.toDot(network));
    }

    return IntrinsicMapTaskExecutor.withSharedCounterSet(
        topoSortedOperations, counterSet, executionContext.getExecutionStateTracker());
  }
@Test
  public void testExecutionContextPlumbing() throws Exception {
    List<ParallelInstruction> instructions =
        Arrays.asList(
            createReadInstruction("Read", ReaderFactoryTest.SingletonTestReaderFactory.class),
            createParDoInstruction(0, 0, "DoFn1", "DoFnUserName"),
            createParDoInstruction(1, 0, "DoFnWithContext", "DoFnWithContextUserName"));

    MapTask mapTask = new MapTask();
    mapTask.setStageName(STAGE);
    mapTask.setInstructions(instructions);
    mapTask.setFactory(Transport.getJsonFactory());

    BatchModeExecutionContext context =
        BatchModeExecutionContext.forTesting(options, counterSet, "testStage");

    try (DataflowMapTaskExecutor executor =
        mapTaskExecutorFactory.create(
            mapTaskToNetwork.apply(mapTask),
            options,
            STAGE,
            readerRegistry,
            sinkRegistry,
            context,
            counterSet,
            idGenerator)) {
      executor.execute();
    }

    List<String> stepNames = new ArrayList<>();
    for (BatchModeExecutionContext.StepContext stepContext : context.getAllStepContexts()) {
      stepNames.add(stepContext.getNameContext().systemName());
    }
    assertThat(stepNames, hasItems("DoFn1", "DoFnWithContext"));
  }