@VisibleForTesting
  static StreamExecutionEnvironment createStreamExecutionEnvironment(FlinkPipelineOptions options) {
    return createStreamExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldAutoSetIdleSourcesFlagWithCheckpointing() {
    // Checkpointing is enabled, never shut down sources
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setCheckpointingInterval(1000L);
    FlinkExecutionEnvironments.createStreamExecutionEnvironment(options);
    assertThat(options.getShutdownSourcesAfterIdleMs(), is(Long.MAX_VALUE));
  }