public static ExecutionEnvironment createBatchExecutionEnvironment(FlinkPipelineOptions options) {
    return createBatchExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldInferParallelismFromEnvironmentBatch() throws IOException {
    String flinkConfDir = extractFlinkConfig();

    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setRunner(TestFlinkRunner.class);
    options.setFlinkMaster("host:80");

    ExecutionEnvironment bev =
        FlinkExecutionEnvironments.createBatchExecutionEnvironment(
            options, Collections.emptyList(), flinkConfDir);

    assertThat(options.getParallelism(), is(23));
    assertThat(bev.getParallelism(), is(23));
  }