@VisibleForTesting
  static StreamExecutionEnvironment createStreamExecutionEnvironment(FlinkPipelineOptions options) {
    return createStreamExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldAutoSetIdleSourcesFlagWithoutCheckpointing() {
    // Checkpointing disabled, shut down sources immediately
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    FlinkExecutionEnvironments.createStreamExecutionEnvironment(options);
    assertThat(options.getShutdownSourcesAfterIdleMs(), is(0L));
  }