public static SourceOperationExecutor create(
      PipelineOptions options,
      SourceOperationRequest request,
      CounterSet counters,
      DataflowExecutionContext<?> executionContext,
      String stageName)
      throws Exception {
    Preconditions.checkNotNull(request, "SourceOperationRequest must be non-null");
    Preconditions.checkNotNull(executionContext, "executionContext must be non-null");

    DataflowOperationContext operationContext =
        executionContext.createOperationContext(
            NameContext.create(
                stageName, request.getOriginalName(), request.getSystemName(), request.getName()));

    return new WorkerCustomSourceOperationExecutor(
        options, request, counters, executionContext, operationContext);
  }
@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));
  }