[FLINK-9045][REST] Make createLocalEnvironmentWithWebUI more user-friendly logging message for web UI address
-add back known logging mesages about webUI address -do not set random port in local stream environment This closes #5814. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/27be32e8 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/27be32e8 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/27be32e8 Branch: refs/heads/master Commit: 27be32e8a44e3afcce9a17e3b95767869f56ab61 Parents: 4f73c8d Author: zentol <ches...@apache.org> Authored: Wed Apr 4 10:44:59 2018 +0200 Committer: zentol <ches...@apache.org> Committed: Mon Apr 16 21:17:54 2018 +0200 ---------------------------------------------------------------------- .../org/apache/flink/api/java/ExecutionEnvironment.java | 6 ++++++ .../flink/runtime/webmonitor/WebMonitorEndpoint.java | 10 +++++++++- .../streaming/api/environment/LocalStreamEnvironment.java | 4 +++- .../api/environment/StreamExecutionEnvironment.java | 6 ++++++ 4 files changed, 24 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/27be32e8/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java ---------------------------------------------------------------------- diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java index 1ce2221..3ea99ea 100644 --- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java +++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java @@ -52,6 +52,7 @@ import org.apache.flink.api.java.typeutils.ValueTypeInfo; import org.apache.flink.api.java.typeutils.runtime.kryo.Serializers; import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.RestOptions; import org.apache.flink.core.fs.Path; import org.apache.flink.types.StringValue; import org.apache.flink.util.NumberSequenceIterator; @@ -1125,6 +1126,11 @@ public abstract class ExecutionEnvironment { conf.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true); + if (!conf.contains(RestOptions.REST_PORT)) { + // explicitly set this option so that it's not set to 0 later + conf.setInteger(RestOptions.REST_PORT, RestOptions.REST_PORT.defaultValue()); + } + return createLocalEnvironment(conf, -1); } http://git-wip-us.apache.org/repos/asf/flink/blob/27be32e8/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java index fb663ad..0ea7550 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java @@ -163,6 +163,8 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp private final FatalErrorHandler fatalErrorHandler; + private boolean hasWebUI = false; + public WebMonitorEndpoint( RestServerEndpointConfiguration endpointConfiguration, GatewayRetriever<? extends T> leaderRetriever, @@ -606,7 +608,10 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp handlers.add(Tuple2.of(YarnStopJobTerminationHeaders.getInstance(), jobStopTerminationHandler)); optWebContent.ifPresent( - webContent -> handlers.add(Tuple2.of(WebContentHandlerSpecification.getInstance(), webContent))); + webContent -> { + handlers.add(Tuple2.of(WebContentHandlerSpecification.getInstance(), webContent)); + hasWebUI = true; + }); // load the log and stdout file handler for the main cluster component final WebMonitorUtils.LogFileLocation logFileLocation = WebMonitorUtils.LogFileLocation.find(clusterConfiguration); @@ -679,6 +684,9 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp @Override public void startInternal() throws Exception { leaderElectionService.start(this); + if (hasWebUI) { + log.info("Web frontend listening at {}.", getRestBaseUrl()); + } } @Override http://git-wip-us.apache.org/repos/asf/flink/blob/27be32e8/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java index 935c78e..b9c76b2 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/LocalStreamEnvironment.java @@ -99,7 +99,9 @@ public class LocalStreamEnvironment extends StreamExecutionEnvironment { // add (and override) the settings with what the user defined configuration.addAll(this.configuration); - configuration.setInteger(RestOptions.REST_PORT, 0); + if (!configuration.contains(RestOptions.REST_PORT)) { + configuration.setInteger(RestOptions.REST_PORT, 0); + } MiniClusterConfiguration cfg = new MiniClusterConfiguration.Builder() .setConfiguration(configuration) http://git-wip-us.apache.org/repos/asf/flink/blob/27be32e8/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java index fa81c27..7372fe8 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java @@ -48,6 +48,7 @@ import org.apache.flink.client.program.PreviewPlanEnvironment; import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.CoreOptions; +import org.apache.flink.configuration.RestOptions; import org.apache.flink.core.fs.Path; import org.apache.flink.runtime.state.AbstractStateBackend; import org.apache.flink.runtime.state.KeyGroupRangeAssignment; @@ -1679,6 +1680,11 @@ public abstract class StreamExecutionEnvironment { conf.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true); + if (!conf.contains(RestOptions.REST_PORT)) { + // explicitly set this option so that it's not set to 0 later + conf.setInteger(RestOptions.REST_PORT, RestOptions.REST_PORT.defaultValue()); + } + return createLocalEnvironment(defaultLocalParallelism, conf); }
