@VisibleForTesting
  static StreamExecutionEnvironment createStreamExecutionEnvironment(FlinkPipelineOptions options) {
    return createStreamExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldFailOnUnknownStateBackend() {
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setStreaming(true);
    options.setStateBackend("unknown");
    options.setStateBackendStoragePath("/path");

    assertThrows(
        "State backend was set to 'unknown' but no storage path was provided.",
        IllegalArgumentException.class,
        () -> FlinkExecutionEnvironments.createStreamExecutionEnvironment(options));
  }