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 testTranslationModeOverrideWithUnboundedSources() {
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setRunner(FlinkRunner.class);
    options.setStreaming(false);

    FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options);
    Pipeline pipeline = Pipeline.create(options);
    pipeline.apply(GenerateSequence.from(0));
    flinkEnv.translate(pipeline);

    assertThat(options.isStreaming(), Matchers.is(true));
  }