public void translate(Pipeline pipeline) {
    this.flinkBatchEnv = null;
    this.flinkStreamEnv = null;

    final boolean hasUnboundedOutput =
        PipelineTranslationModeOptimizer.hasUnboundedOutput(pipeline);
    if (hasUnboundedOutput) {
      LOG.info("Found unbounded PCollection. Switching to streaming execution.");
      options.setStreaming(true);
    }

    // Staged files need to be set before initializing the execution environments
    prepareFilesToStageForRemoteClusterExecution(options);

    FlinkPipelineTranslator translator;
    if (options.isStreaming() || options.getUseDataStreamForBatch()) {
      this.flinkStreamEnv = FlinkExecutionEnvironments.createStreamExecutionEnvironment(options);
      if (hasUnboundedOutput && !flinkStreamEnv.getCheckpointConfig().isCheckpointingEnabled()) {
        LOG.warn(
            "UnboundedSources present which rely on checkpointing, but checkpointing is disabled.");
      }
      translator =
          new FlinkStreamingPipelineTranslator(flinkStreamEnv, options, options.isStreaming());
      if (!options.isStreaming()) {
        flinkStreamEnv.setRuntimeMode(RuntimeExecutionMode.BATCH);
      }
    } else {
      this.flinkBatchEnv = FlinkExecutionEnvironments.createBatchExecutionEnvironment(options);
      translator = new FlinkBatchPipelineTranslator(flinkBatchEnv, options);
    }

    // Transform replacements need to receive the finalized PipelineOptions
    // including execution mode (batch/streaming) and parallelism.
    pipeline.replaceAll(FlinkTransformOverrides.getDefaultOverrides(options));

    translator.translate(pipeline);
  }
@Test
  public void shouldProvideParallelismToTransformOverrides() {
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setStreaming(true);
    options.setRunner(FlinkRunner.class);
    FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options);
    Pipeline p = Pipeline.create(options);
    // Create a transform applicable for PTransformMatchers.writeWithRunnerDeterminedSharding()
    // which requires parallelism
    p.apply(Create.of("test")).apply(TextIO.write().to("/tmp"));
    p = Mockito.spy(p);

    // If this succeeds we're ok
    flinkEnv.translate(p);

    // Verify we were using desired replacement transform
    ArgumentCaptor<ImmutableList> captor = ArgumentCaptor.forClass(ImmutableList.class);
    Mockito.verify(p).replaceAll(captor.capture());
    ImmutableList<PTransformOverride> overridesList = captor.getValue();
    assertThat(
        overridesList,
        hasItem(
            new BaseMatcher<PTransformOverride>() {
              @Override
              public void describeTo(Description description) {}

              @Override
              public boolean matches(Object actual) {
                if (actual instanceof PTransformOverride) {
                  PTransformOverrideFactory overrideFactory =
                      ((PTransformOverride) actual).getOverrideFactory();
                  if (overrideFactory
                      instanceof FlinkStreamingPipelineTranslator.StreamingShardedWriteFactory) {
                    FlinkStreamingPipelineTranslator.StreamingShardedWriteFactory factory =
                        (FlinkStreamingPipelineTranslator.StreamingShardedWriteFactory)
                            overrideFactory;
                    return factory.options.getParallelism() > 0;
                  }
                }
                return false;
              }
            }));
  }