IntrinsicMapTaskExecutorFactory implements DataflowMapTaskExecutorFactory { OperationNode createWriteOperation( ParallelInstructionNode node, PipelineOptions options, SinkFactory sinkFactory, DataflowExecutionContext executionContext, DataflowOperationContext context) throws Exception { ParallelInstruction instruction = node.getParallelInstruction(); WriteInstruction write = instruction.getWrite(); Coder<?> coder = CloudObjects.coderFromCloudObject(CloudObject.fromSpec(write.getSink().getCodec())); CloudObject cloudSink = CloudObject.fromSpec(write.getSink().getSpec()); Sink<?> sink = sinkFactory.create(cloudSink, coder, options, executionContext, context); return OperationNode.create(WriteOperation.create(sink, EMPTY_OUTPUT_RECEIVER_ARRAY, 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);  }
@SuppressWarnings("unchecked") @Test public void testCreateWriteOperation() throws Exception { int producerIndex = 1; int producerOutputNum = 2; ParallelInstructionNode instructionNode = ParallelInstructionNode.create( createWriteInstruction(producerIndex, producerOutputNum, "WriteOperation"), ExecutionLocation.UNKNOWN); 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(WriteOperation.class)); WriteOperation writeOperation = (WriteOperation) ((OperationNode) operationNode).getOperation(); assertEquals(0, writeOperation.receivers.length); assertEquals(Operation.InitializationState.UNSTARTED, writeOperation.initializationState); assertThat(writeOperation.sink, instanceOf(SizeReportingSinkWrapper.class)); assertThat( ((SizeReportingSinkWrapper<?>) writeOperation.sink).getUnderlyingSink(), instanceOf(TestSink.class)); counterSet.extractUpdates(false, updateExtractor); verify(updateExtractor).longSum(eq(named("WriteOperation-ByteCount")), anyBoolean(), anyLong()); verifyNoMoreInteractions(updateExtractor); }