@VisibleForTesting
  static StreamExecutionEnvironment createStreamExecutionEnvironment(FlinkPipelineOptions options) {
    return createStreamExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldSetWebUIOptions() {
    PipelineOptionsFactory.register(TestOptions.class);
    PipelineOptionsFactory.register(FlinkPipelineOptions.class);

    FlinkPipelineOptions options =
        PipelineOptionsFactory.fromArgs(
                "--key1=value1",
                "--key2",
                "--key3=",
                "--parallelism=10",
                "--checkpointTimeoutMillis=500")
            .as(FlinkPipelineOptions.class);

    StreamExecutionEnvironment sev =
        FlinkExecutionEnvironments.createStreamExecutionEnvironment(options);

    Map<String, String> actualMap = sev.getConfig().getGlobalJobParameters().toMap();

    Map<String, String> expectedMap = new HashMap<>();
    expectedMap.put("key1", "value1");
    expectedMap.put("key2", "true");
    expectedMap.put("key3", "");
    expectedMap.put("checkpointTimeoutMillis", "500");
    expectedMap.put("parallelism", "10");

    Map<String, String> filteredMap =
        expectedMap.entrySet().stream()
            .filter(
                kv ->
                    actualMap.containsKey(kv.getKey())
                        && kv.getValue().equals(actualMap.get(kv.getKey())))
            .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue()));

    assertTrue(expectedMap.size() == filteredMap.size());
  }