IntrinsicMapTaskExecutorFactory implements DataflowMapTaskExecutorFactory { OperationNode createFlattenOperation( Network<Node, Edge> network, ParallelInstructionNode node, OperationContext context) { OutputReceiver[] receivers = getOutputReceivers(network, node); return OperationNode.create(new FlattenOperation(receivers, context)); } 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 testCreateFlattenOperation() throws Exception { int producerIndex1 = 1; int producerOutputNum1 = 2; int producerIndex2 = 0; int producerOutputNum2 = 1; ParallelInstructionNode instructionNode = ParallelInstructionNode.create( createFlattenInstruction( producerIndex1, producerOutputNum1, producerIndex2, producerOutputNum2, "Flatten"), ExecutionLocation.UNKNOWN); when(network.successors(instructionNode)) .thenReturn( ImmutableSet.<Node>of( IntrinsicMapTaskExecutorFactory.createOutputReceiversTransform(STAGE, counterSet) .apply( InstructionOutputNode.create( instructionNode.getParallelInstruction().getOutputs().get(0), PCOLLECTION_ID)))); when(network.outDegree(instructionNode)).thenReturn(1); Node operationNode = mapTaskExecutorFactory .createOperationTransformForParallelInstructionNodes( STAGE, network, options, readerRegistry, sinkRegistry, BatchModeExecutionContext.forTesting(options, counterSet, "testStage")) .apply(instructionNode); assertThat(operationNode, instanceOf(OperationNode.class)); assertThat(((OperationNode) operationNode).getOperation(), instanceOf(FlattenOperation.class)); FlattenOperation flattenOperation = (FlattenOperation) ((OperationNode) operationNode).getOperation(); assertEquals(1, flattenOperation.receivers.length); assertEquals(0, flattenOperation.receivers[0].getReceiverCount()); assertEquals(Operation.InitializationState.UNSTARTED, flattenOperation.initializationState); }