FlinkExecutionEnvironments { public static StreamExecutionEnvironment createStreamExecutionEnvironment( FlinkPipelineOptions options, List<String> filesToStage) { return createStreamExecutionEnvironment(options, filesToStage, null); }  static ExecutionEnvironment createBatchExecutionEnvironment(
      FlinkPipelineOptions options, List<String> filesToStage); static StreamExecutionEnvironment createStreamExecutionEnvironment(
      FlinkPipelineOptions options, List<String> filesToStage);  }
@Test public void shouldSetParallelismStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setParallelism(42); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(options.getParallelism(), is(42)); assertThat(sev.getParallelism(), is(42)); }
@Test public void shouldSetMaxParallelismStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setMaxParallelism(42); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(options.getMaxParallelism(), is(42)); assertThat(sev.getMaxParallelism(), is(42)); }
@Test public void shouldInferParallelismFromEnvironmentStreaming() throws IOException { String confDir = extractFlinkConfig(); FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setFlinkMaster("host:80"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList(), confDir); assertThat(options.getParallelism(), is(23)); assertThat(sev.getParallelism(), is(23)); }
@Test public void shouldFallbackToDefaultParallelismStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setFlinkMaster("host:80"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(options.getParallelism(), is(1)); assertThat(sev.getParallelism(), is(1)); }
@Test public void useDefaultParallelismFromContextStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(sev, instanceOf(LocalStreamEnvironment.class)); assertThat(options.getParallelism(), is(LocalStreamEnvironment.getDefaultLocalParallelism())); assertThat(sev.getParallelism(), is(LocalStreamEnvironment.getDefaultLocalParallelism())); }
@Test public void shouldParsePortForRemoteEnvironmentStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host:1234"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(sev, instanceOf(RemoteStreamEnvironment.class)); checkHostAndPort(sev, "host", 1234); }
@Test public void shouldAllowPortOmissionForRemoteEnvironmentStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(sev, instanceOf(RemoteStreamEnvironment.class)); checkHostAndPort(sev, "host", RestOptions.PORT.defaultValue()); }
@Test public void shouldTreatAutoAndEmptyHostTheSameStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); options.setFlinkMaster("[auto]"); StreamExecutionEnvironment sev2 = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertEquals(sev.getClass(), sev2.getClass()); }
@Test public void shouldDetectMalformedPortStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host:p0rt"); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("Unparseable port number"); FlinkExecutionEnvironments.createStreamExecutionEnvironment(options, Collections.emptyList()); }
@Test public void shouldSupportIPv6Streaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("[FE80:CD00:0000:0CDE:1257:0000:211E:729C]:1234"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(sev, "FE80:CD00:0000:0CDE:1257:0000:211E:729C", 1234); options.setFlinkMaster("FE80:CD00:0000:0CDE:1257:0000:211E:729C"); sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort( sev, "FE80:CD00:0000:0CDE:1257:0000:211E:729C", RestOptions.PORT.defaultValue()); }
@Test public void shouldRemoveHttpProtocolFromHostStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); for (String flinkMaster : new String[] { "http: }) { options.setFlinkMaster(flinkMaster); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); checkHostAndPort(sev, "host", 1234); } }
@Test public void shouldAutoSetIdleSourcesFlagWithoutCheckpointing() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); FlinkExecutionEnvironments.createStreamExecutionEnvironment(options, Collections.emptyList()); assertThat(options.getShutdownSourcesAfterIdleMs(), is(0L)); }
@Test public void shouldAutoSetIdleSourcesFlagWithCheckpointing() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setCheckpointingInterval(1000L); FlinkExecutionEnvironments.createStreamExecutionEnvironment(options, Collections.emptyList()); assertThat(options.getShutdownSourcesAfterIdleMs(), is(Long.MAX_VALUE)); }
@Test public void shouldAcceptExplicitlySetIdleSourcesFlagWithoutCheckpointing() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setShutdownSourcesAfterIdleMs(42L); FlinkExecutionEnvironments.createStreamExecutionEnvironment(options, Collections.emptyList()); assertThat(options.getShutdownSourcesAfterIdleMs(), is(42L)); }
@Test public void shouldAcceptExplicitlySetIdleSourcesFlagWithCheckpointing() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setCheckpointingInterval(1000L); options.setShutdownSourcesAfterIdleMs(42L); FlinkExecutionEnvironments.createStreamExecutionEnvironment(options, Collections.emptyList()); assertThat(options.getShutdownSourcesAfterIdleMs(), is(42L)); }
@Test public void shouldSetSavepointRestoreForRemoteStreaming() { String path = "fakePath"; FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setFlinkMaster("host:80"); options.setSavepointPath(path); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(sev, instanceOf(RemoteStreamEnvironment.class)); assertThat(getSavepointPath(sev), is(path)); }
@Test public void shouldParsePortForRemoteEnvironmentStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host:1234"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(sev, instanceOf(RemoteStreamEnvironment.class)); assertThat(Whitebox.getInternalState(sev, "host"), is("host")); assertThat(Whitebox.getInternalState(sev, "port"), is(1234)); }
@Test public void shouldAllowPortOmissionForRemoteEnvironmentStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("host"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(sev, instanceOf(RemoteStreamEnvironment.class)); assertThat(Whitebox.getInternalState(sev, "host"), is("host")); assertThat(Whitebox.getInternalState(sev, "port"), is(RestOptions.PORT.defaultValue())); }
@Test public void shouldSupportIPv6Streaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setFlinkMaster("[FE80:CD00:0000:0CDE:1257:0000:211E:729C]:1234"); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat( Whitebox.getInternalState(sev, "host"), is("FE80:CD00:0000:0CDE:1257:0000:211E:729C")); assertThat(Whitebox.getInternalState(sev, "port"), is(1234)); options.setFlinkMaster("FE80:CD00:0000:0CDE:1257:0000:211E:729C"); sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat( Whitebox.getInternalState(sev, "host"), is("FE80:CD00:0000:0CDE:1257:0000:211E:729C")); assertThat(Whitebox.getInternalState(sev, "port"), is(RestOptions.PORT.defaultValue())); }
@Test public void shouldRemoveHttpProtocolFromHostStreaming() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); for (String flinkMaster : new String[] { "http: }) { options.setFlinkMaster(flinkMaster); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(Whitebox.getInternalState(sev, "host"), is("host")); assertThat(Whitebox.getInternalState(sev, "port"), is(1234)); } }
@Test public void shouldSetSavepointRestoreForRemoteStreaming() { String path = "fakePath"; FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setFlinkMaster("host:80"); options.setSavepointPath(path); StreamExecutionEnvironment sev = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, Collections.emptyList()); assertThat(sev, instanceOf(RemoteStreamEnvironment.class)); assertThat( Whitebox.getInternalState(sev, "restoreSettings"), is(SavepointRestoreSettings.forPath(path))); }