@VisibleForTesting
  static StreamExecutionEnvironment createStreamExecutionEnvironment(FlinkPipelineOptions options) {
    return createStreamExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldCreateRocksDbStateBackend() {
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setStreaming(true);
    options.setStateBackend("rocksDB");
    options.setStateBackendStoragePath(temporaryFolder.getRoot().toURI().toString());

    StreamExecutionEnvironment sev =
        FlinkExecutionEnvironments.createStreamExecutionEnvironment(options);

    assertThat(sev.getStateBackend(), instanceOf(RocksDBStateBackend.class));
  }