FlinkExecutionEnvironments { public static ExecutionEnvironment createBatchExecutionEnvironment( FlinkPipelineOptions options, List<String> filesToStage) { return createBatchExecutionEnvironment(options, filesToStage, null); }  static ExecutionEnvironment createBatchExecutionEnvironment(
      FlinkPipelineOptions options, List<String> filesToStage); static StreamExecutionEnvironment createStreamExecutionEnvironment(
      FlinkPipelineOptions options, List<String> filesToStage);  }
@Test public void shouldSetParallelismBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setParallelism(42); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(options.getParallelism(), is(42)); assertThat(bev.getParallelism(), is(42)); }
@Test public void shouldInferParallelismFromEnvironmentBatch() throws IOException { String flinkConfDir = extractFlinkConfig(); FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); 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)); }
@Test public void shouldFallbackToDefaultParallelismBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setFlinkMaster("host:80"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(options.getParallelism(), is(1)); assertThat(bev.getParallelism(), is(1)); }
@Test public void useDefaultParallelismFromContextBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(bev, instanceOf(LocalEnvironment.class)); assertThat(options.getParallelism(), is(LocalStreamEnvironment.getDefaultLocalParallelism())); assertThat(bev.getParallelism(), is(LocalStreamEnvironment.getDefaultLocalParallelism())); }
@Test public void shouldParsePortForRemoteEnvironmentBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(bev, instanceOf(RemoteEnvironment.class)); checkHostAndPort(bev, "host", 1234); }
@Test public void shouldAllowPortOmissionForRemoteEnvironmentBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(bev, instanceOf(RemoteEnvironment.class)); checkHostAndPort(bev, "host", RestOptions.PORT.defaultValue()); }
@Test public void shouldTreatAutoAndEmptyHostTheSameBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); ExecutionEnvironment sev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); options.setFlinkMaster("[auto]"); ExecutionEnvironment sev2 = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertEquals(sev.getClass(), sev2.getClass()); }
@Test public void shouldDetectMalformedPortBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host:p0rt"); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("Unparseable port number"); FlinkExecutionEnvironments.createBatchExecutionEnvironment(options, Collections.emptyList()); }
@Test public void shouldSupportIPv4Batch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("192.168.1.1:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(bev, "192.168.1.1", 1234); options.setFlinkMaster("192.168.1.1"); bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(bev, "192.168.1.1", RestOptions.PORT.defaultValue()); }
@Test public void shouldSupportIPv4Streaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("192.168.1.1:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(bev, "192.168.1.1", 1234); options.setFlinkMaster("192.168.1.1"); bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(bev, "192.168.1.1", RestOptions.PORT.defaultValue()); }
@Test public void shouldSupportIPv6Batch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("[FE80:CD00:0000:0CDE:1257:0000:211E:729C]:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(bev, "FE80:CD00:0000:0CDE:1257:0000:211E:729C", 1234); options.setFlinkMaster("FE80:CD00:0000:0CDE:1257:0000:211E:729C"); bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort( bev, "FE80:CD00:0000:0CDE:1257:0000:211E:729C", RestOptions.PORT.defaultValue()); }
@Test public void shouldRemoveHttpProtocolFromHostBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); for (String flinkMaster : new String[] { "http: }) { options.setFlinkMaster(flinkMaster); ExecutionEnvironment sev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(sev, "host", 1234); } }
@Test public void shouldParsePortForRemoteEnvironmentBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(bev, instanceOf(RemoteEnvironment.class)); assertThat(Whitebox.getInternalState(bev, "host"), is("host")); assertThat(Whitebox.getInternalState(bev, "port"), is(1234)); }
@Test public void shouldAllowPortOmissionForRemoteEnvironmentBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(bev, instanceOf(RemoteEnvironment.class)); assertThat(Whitebox.getInternalState(bev, "host"), is("host")); assertThat(Whitebox.getInternalState(bev, "port"), is(RestOptions.PORT.defaultValue())); }
@Test public void shouldSupportIPv4Batch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("192.168.1.1:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(Whitebox.getInternalState(bev, "host"), is("192.168.1.1")); assertThat(Whitebox.getInternalState(bev, "port"), is(1234)); options.setFlinkMaster("192.168.1.1"); bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(Whitebox.getInternalState(bev, "host"), is("192.168.1.1")); assertThat(Whitebox.getInternalState(bev, "port"), is(RestOptions.PORT.defaultValue())); }
@Test public void shouldSupportIPv4Streaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("192.168.1.1:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(Whitebox.getInternalState(bev, "host"), is("192.168.1.1")); assertThat(Whitebox.getInternalState(bev, "port"), is(1234)); options.setFlinkMaster("192.168.1.1"); bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(Whitebox.getInternalState(bev, "host"), is("192.168.1.1")); assertThat(Whitebox.getInternalState(bev, "port"), is(RestOptions.PORT.defaultValue())); }
@Test public void shouldSupportIPv6Batch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("[FE80:CD00:0000:0CDE:1257:0000:211E:729C]:1234"); ExecutionEnvironment bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat( Whitebox.getInternalState(bev, "host"), is("FE80:CD00:0000:0CDE:1257:0000:211E:729C")); assertThat(Whitebox.getInternalState(bev, "port"), is(1234)); options.setFlinkMaster("FE80:CD00:0000:0CDE:1257:0000:211E:729C"); bev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat( Whitebox.getInternalState(bev, "host"), is("FE80:CD00:0000:0CDE:1257:0000:211E:729C")); assertThat(Whitebox.getInternalState(bev, "port"), is(RestOptions.PORT.defaultValue())); }
@Test public void shouldRemoveHttpProtocolFromHostBatch() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); for (String flinkMaster : new String[] { "http: }) { options.setFlinkMaster(flinkMaster); ExecutionEnvironment sev = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, Collections.emptyList()); assertThat(Whitebox.getInternalState(sev, "host"), is("host")); assertThat(Whitebox.getInternalState(sev, "port"), is(1234)); } }