public static ExecutionEnvironment createBatchExecutionEnvironment(FlinkPipelineOptions options) {
    return createBatchExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void useDefaultParallelismFromContextBatch() {
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setRunner(TestFlinkRunner.class);

    ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment(options);

    assertThat(bev, instanceOf(LocalEnvironment.class));
    assertThat(options.getParallelism(), is(LocalStreamEnvironment.getDefaultLocalParallelism()));
    assertThat(bev.getParallelism(), is(LocalStreamEnvironment.getDefaultLocalParallelism()));
  }