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));
  }
@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);
  }