@VisibleForTesting
  static StreamExecutionEnvironment createStreamExecutionEnvironment(FlinkPipelineOptions options) {
    return createStreamExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldSetSavepointRestoreForRemoteStreaming() {
    String path = "fakePath";
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setRunner(TestFlinkRunner.class);
    options.setFlinkMaster("host:80");
    options.setSavepointPath(path);

    StreamExecutionEnvironment sev =
        FlinkExecutionEnvironments.createStreamExecutionEnvironment(options);
    // subject to change with https://issues.apache.org/jira/browse/FLINK-11048
    assertThat(sev, instanceOf(RemoteStreamEnvironment.class));
    assertThat(getSavepointPath(sev), is(path));
  }