@VisibleForTesting
  static StreamExecutionEnvironment createStreamExecutionEnvironment(FlinkPipelineOptions options) {
    return createStreamExecutionEnvironment(
        options,
        MoreObjects.firstNonNull(options.getFilesToStage(), Collections.emptyList()),
        options.getFlinkConfDir());
  }
@Test
  public void shouldDetectMalformedPortStreaming() {
    FlinkPipelineOptions options = getDefaultPipelineOptions();
    options.setRunner(FlinkRunner.class);
    options.setFlinkMaster("host:p0rt");

    expectedException.expect(IllegalArgumentException.class);
    expectedException.expectMessage("Unparseable port number");

    FlinkExecutionEnvironments.createStreamExecutionEnvironment(options);
  }