@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 testCreateMapTaskExecutor() throws Exception {
    List<ParallelInstruction> instructions =
        Arrays.asList(
            createReadInstruction("Read"),
            createParDoInstruction(0, 0, "DoFn1"),
            createParDoInstruction(0, 0, "DoFnWithContext"),
            createFlattenInstruction(1, 0, 2, 0, "Flatten"),
            createWriteInstruction(3, 0, "Write"));

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

    try (DataflowMapTaskExecutor executor =
        mapTaskExecutorFactory.create(
            mapTaskToNetwork.apply(mapTask),
            options,
            STAGE,
            readerRegistry,
            sinkRegistry,
            BatchModeExecutionContext.forTesting(options, counterSet, "testStage"),
            counterSet,
            idGenerator)) {
      // Safe covariant cast not expressible without rawtypes.
      @SuppressWarnings({
        "rawtypes", // TODO(https://github.com/apache/beam/issues/20447)
        "unchecked"
      })
      List<Object> operations = (List) executor.operations;
      assertThat(
          operations,
          hasItems(
              instanceOf(ReadOperation.class),
              instanceOf(ParDoOperation.class),
              instanceOf(ParDoOperation.class),
              instanceOf(FlattenOperation.class),
              instanceOf(WriteOperation.class)));

      // Verify that the inputs are attached.
      ReadOperation readOperation =
          Iterables.getOnlyElement(Iterables.filter(operations, ReadOperation.class));
      assertEquals(2, readOperation.receivers[0].getReceiverCount());

      FlattenOperation flattenOperation =
          Iterables.getOnlyElement(Iterables.filter(operations, FlattenOperation.class));
      for (ParDoOperation operation : Iterables.filter(operations, ParDoOperation.class)) {
        assertSame(flattenOperation, operation.receivers[0].getOnlyReceiver());
      }
      WriteOperation writeOperation =
          Iterables.getOnlyElement(Iterables.filter(operations, WriteOperation.class));
      assertSame(writeOperation, flattenOperation.receivers[0].getOnlyReceiver());
    }

    @SuppressWarnings("unchecked")
    Counter<Long, ?> otherMsecCounter =
        (Counter<Long, ?>) counterSet.getExistingCounter("test-other-msecs");

    // "other" state only got created upon MapTaskExecutor.execute().
    assertNull(otherMsecCounter);

    counterSet.extractUpdates(false, updateExtractor);
    verifyOutputCounters(
        updateExtractor,
        "read_output_name",
        "DoFn1_output",
        "DoFnWithContext_output",
        "flatten_output_name");
    verify(updateExtractor).longSum(eq(named("Read-ByteCount")), anyBoolean(), anyLong());
    verify(updateExtractor).longSum(eq(named("Write-ByteCount")), anyBoolean(), anyLong());
    verifyNoMoreInteractions(updateExtractor);
  }