SourceOperationExecutorFactory { public static SourceOperationExecutor create( PipelineOptions options, SourceOperationRequest request, CounterSet counters, DataflowExecutionContext<?> executionContext, String stageName) throws Exception { boolean beamFnApi = DataflowRunner.hasExperiment(options.as(DataflowPipelineDebugOptions.class), "beam_fn_api"); Preconditions.checkNotNull(request, "SourceOperationRequest must be non-null"); Preconditions.checkNotNull(executionContext, "executionContext must be non-null"); if (beamFnApi) { return new NoOpSourceOperationExecutor(request); } else { DataflowOperationContext operationContext = executionContext.createOperationContext( NameContext.create( stageName, request.getOriginalName(), request.getSystemName(), request.getName())); return new WorkerCustomSourceOperationExecutor( options, request, counters, executionContext, operationContext); } }  static SourceOperationExecutor create(
      PipelineOptions options,
      SourceOperationRequest request,
      CounterSet counters,
      DataflowExecutionContext<?> executionContext,
      String stageName);  }
@Test public void testCreateDefault() throws Exception { SourceOperationRequest request = new SourceOperationRequest() .setName("name") .setOriginalName("original") .setSystemName("system") .setStageName("stage") .setSplit(new SourceSplitRequest()); DataflowOperationContext mockOperation = Mockito.mock(DataflowOperationContext.class); Mockito.when(executionContext.createOperationContext(Mockito.isA(NameContext.class))) .thenReturn(mockOperation); SourceOperationExecutor sourceOperationExecutor = SourceOperationExecutorFactory.create( PipelineOptionsFactory.create(), request, null, executionContext, "STAGE"); assertThat(sourceOperationExecutor, instanceOf(WorkerCustomSourceOperationExecutor.class)); }
@Test public void testCreateWithFnApi() throws Exception { SourceOperationRequest request = new SourceOperationRequest() .setName("name") .setOriginalName("original") .setSystemName("system") .setStageName("stage"); PipelineOptions options = PipelineOptionsFactory.fromArgs(new String[] {"--experiments=beam_fn_api"}).create(); assertThat( SourceOperationExecutorFactory.create(options, request, null, executionContext, "STAGE"), instanceOf(NoOpSourceOperationExecutor.class)); }