[FLINK-10101][mesos] Announce web UI url at Mesos master

This commit announces the cluster's web UI url at the Mesos master so that it
is visible in the Mesos UI.

This closes apache#6522.

flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java

@@ -145,8 +145,8 @@ protected ResourceManager<?> createResourceManager(
 			mesosServices,
 			schedulerConfiguration,
 			taskManagerParameters,
-			taskManagerContainerSpec
-			);
+			taskManagerContainerSpec,
+			webInterfaceUrl);
 	}
 
 	@Override

flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java

@@ -135,8 +135,8 @@ protected ResourceManager<?> createResourceManager(
 			mesosServices,
 			mesosConfig,
 			taskManagerParameters,
-			taskManagerContainerSpec
-			);
+			taskManagerContainerSpec,
+			webInterfaceUrl);
 	}
 
 	@Override

flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java

@@ -120,6 +120,10 @@ public class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerN
 	/** A local actor system for using the helper actors. */
 	private final ActorSystem actorSystem;
 
+	/** Web url to show in mesos page. */
+	@Nullable
+	private final String webUiUrl;
+
 	/** Mesos scheduler driver. */
 	private SchedulerDriver schedulerDriver;
 
@@ -158,7 +162,8 @@ public MesosResourceManager(
 			MesosServices mesosServices,
 			MesosConfiguration mesosConfig,
 			MesosTaskManagerParameters taskManagerParameters,
-			ContainerSpecification taskManagerContainerSpec) {
+			ContainerSpecification taskManagerContainerSpec,
+			@Nullable String webUiUrl) {
 		super(
 			rpcService,
 			resourceManagerEndpointId,
@@ -181,6 +186,7 @@ public MesosResourceManager(
 
 		this.taskManagerParameters = Preconditions.checkNotNull(taskManagerParameters);
 		this.taskManagerContainerSpec = Preconditions.checkNotNull(taskManagerContainerSpec);
+		this.webUiUrl = webUiUrl;
 
 		this.workersInNew = new HashMap<>(8);
 		this.workersInLaunch = new HashMap<>(8);
@@ -237,6 +243,10 @@ protected void initialize() throws ResourceManagerException {
 		Protos.FrameworkInfo.Builder frameworkInfo = mesosConfig.frameworkInfo()
 			.clone()
 			.setCheckpoint(true);
+		if (webUiUrl != null) {
+			frameworkInfo.setWebuiUrl(webUiUrl);
+		}
+
 		try {
 			Option<Protos.FrameworkID> frameworkID = workerStore.getFrameworkID();
 			if (frameworkID.isEmpty()) {

flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerTest.java

@@ -188,7 +188,8 @@ public TestingMesosResourceManager(
 				mesosServices,
 				mesosConfig,
 				taskManagerParameters,
-				taskManagerContainerSpec);
+				taskManagerContainerSpec,
+				null);
 		}
 
 		@Override