IntrinsicMapTaskExecutorFactory implements DataflowMapTaskExecutorFactory { private OperationNode createParDoOperation( Network<Node, Edge> network, ParallelInstructionNode node, PipelineOptions options, DataflowExecutionContext<?> executionContext, DataflowOperationContext operationContext) throws Exception { ParallelInstruction instruction = node.getParallelInstruction(); ParDoInstruction parDo = instruction.getParDo(); TupleTag<?> mainOutputTag = tupleTag(parDo.getMultiOutputInfos().get(0)); ImmutableMap.Builder<TupleTag<?>, Integer> outputTagsToReceiverIndicesBuilder = ImmutableMap.builder(); int successorOffset = 0; for (Node successor : network.successors(node)) { for (Edge edge : network.edgesConnecting(node, successor)) { outputTagsToReceiverIndicesBuilder.put( tupleTag(((MultiOutputInfoEdge) edge).getMultiOutputInfo()), successorOffset); } successorOffset += 1; } ParDoFn fn = parDoFnFactory.create( options, CloudObject.fromSpec(parDo.getUserFn()), parDo.getSideInputs(), mainOutputTag, outputTagsToReceiverIndicesBuilder.build(), executionContext, operationContext); OutputReceiver[] receivers = getOutputReceivers(network, node); return OperationNode.create(new ParDoOperation(fn, receivers, operationContext)); } private  IntrinsicMapTaskExecutorFactory(); static IntrinsicMapTaskExecutorFactory defaultFactory(); @Override DataflowMapTaskExecutor create(
      InstructionRequestHandler instructionRequestHandler,
      GrpcFnServer<GrpcDataService> grpcDataFnServer,
      Endpoints.ApiServiceDescriptor dataApiServiceDescriptor,
      GrpcFnServer<GrpcStateService> grpcStateFnServer,
      MutableNetwork<Node, Edge> network,
      PipelineOptions options,
      String stageName,
      ReaderFactory readerFactory,
      SinkFactory sinkFactory,
      DataflowExecutionContext<?> executionContext,
      CounterSet counterSet,
      IdGenerator idGenerator);  }
@Test public void testCreateParDoOperation() throws Exception { int producerIndex = 1; int producerOutputNum = 2; BatchModeExecutionContext context = BatchModeExecutionContext.forTesting(options, counterSet, "testStage"); ParallelInstructionNode instructionNode = ParallelInstructionNode.create( createParDoInstruction(producerIndex, producerOutputNum, "DoFn"), ExecutionLocation.UNKNOWN); Node outputReceiverNode = IntrinsicMapTaskExecutorFactory.createOutputReceiversTransform(STAGE, counterSet) .apply( InstructionOutputNode.create( instructionNode.getParallelInstruction().getOutputs().get(0), PCOLLECTION_ID)); when(network.successors(instructionNode)).thenReturn(ImmutableSet.of(outputReceiverNode)); when(network.outDegree(instructionNode)).thenReturn(1); when(network.edgesConnecting(instructionNode, outputReceiverNode)) .thenReturn( ImmutableSet.<Edge>of( MultiOutputInfoEdge.create( instructionNode .getParallelInstruction() .getParDo() .getMultiOutputInfos() .get(0)))); Node operationNode = mapTaskExecutorFactory .createOperationTransformForParallelInstructionNodes( STAGE, network, options, readerRegistry, sinkRegistry, context) .apply(instructionNode); assertThat(operationNode, instanceOf(OperationNode.class)); assertThat(((OperationNode) operationNode).getOperation(), instanceOf(ParDoOperation.class)); ParDoOperation parDoOperation = (ParDoOperation) ((OperationNode) operationNode).getOperation(); assertEquals(1, parDoOperation.receivers.length); assertEquals(0, parDoOperation.receivers[0].getReceiverCount()); assertEquals(Operation.InitializationState.UNSTARTED, parDoOperation.initializationState); }