IntrinsicMapTaskExecutorFactory implements DataflowMapTaskExecutorFactory { OperationNode createReadOperation( Network<Node, Edge> network, ParallelInstructionNode node, PipelineOptions options, ReaderFactory readerFactory, DataflowExecutionContext<?> executionContext, DataflowOperationContext operationContext) throws Exception { ParallelInstruction instruction = node.getParallelInstruction(); ReadInstruction read = instruction.getRead(); Source cloudSource = CloudSourceUtils.flattenBaseSpecs(read.getSource()); CloudObject sourceSpec = CloudObject.fromSpec(cloudSource.getSpec()); Coder<?> coder = CloudObjects.coderFromCloudObject(CloudObject.fromSpec(cloudSource.getCodec())); NativeReader<?> reader = readerFactory.create(sourceSpec, coder, options, executionContext, operationContext); OutputReceiver[] receivers = getOutputReceivers(network, node); return OperationNode.create(ReadOperation.create(reader, 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 testCreateReadOperation() throws Exception { ParallelInstructionNode instructionNode = ParallelInstructionNode.create(createReadInstruction("Read"), 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, PipelineOptionsFactory.create(), readerRegistry, sinkRegistry, BatchModeExecutionContext.forTesting(options, counterSet, "testStage")) .apply(instructionNode); assertThat(operationNode, instanceOf(OperationNode.class)); assertThat(((OperationNode) operationNode).getOperation(), instanceOf(ReadOperation.class)); ReadOperation readOperation = (ReadOperation) ((OperationNode) operationNode).getOperation(); assertEquals(1, readOperation.receivers.length); assertEquals(0, readOperation.receivers[0].getReceiverCount()); assertEquals(Operation.InitializationState.UNSTARTED, readOperation.initializationState); assertThat(readOperation.reader, instanceOf(ReaderFactoryTest.TestReader.class)); counterSet.extractUpdates(false, updateExtractor); verifyOutputCounters(updateExtractor, "read_output_name"); verify(updateExtractor).longSum(eq(named("Read-ByteCount")), anyBoolean(), anyLong()); verifyNoMoreInteractions(updateExtractor); }