[FLINK-8120] [flip6] Register Yarn application with correct tracking URL

The cluster entrypoints start the ResourceManager with the web interface URL.
This URL is used to set the correct tracking URL in Yarn when registering the
Yarn application.

This closes #5128.

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

@@ -53,6 +53,8 @@ import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.PosixParser;
 
+import javax.annotation.Nullable;
+
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
@@ -124,13 +126,14 @@ public class MesosJobClusterEntrypoint extends JobClusterEntrypoint {
 
 	@Override
 	protected ResourceManager<?> createResourceManager(
-		Configuration configuration,
-		ResourceID resourceId,
-		RpcService rpcService,
-		HighAvailabilityServices highAvailabilityServices,
-		HeartbeatServices heartbeatServices,
-		MetricRegistry metricRegistry,
-		FatalErrorHandler fatalErrorHandler) throws Exception {
+			Configuration configuration,
+			ResourceID resourceId,
+			RpcService rpcService,
+			HighAvailabilityServices highAvailabilityServices,
+			HeartbeatServices heartbeatServices,
+			MetricRegistry metricRegistry,
+			FatalErrorHandler fatalErrorHandler,
+			@Nullable String webInterfaceUrl) throws Exception {
 		final ResourceManagerConfiguration rmConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServices rmRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(

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

@@ -51,6 +51,8 @@ import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.PosixParser;
 
+import javax.annotation.Nullable;
+
 /**
  * Entry point for Mesos session clusters.
  */
@@ -114,13 +116,14 @@ public class MesosSessionClusterEntrypoint extends SessionClusterEntrypoint {
 
 	@Override
 	protected ResourceManager<?> createResourceManager(
-		Configuration configuration,
-		ResourceID resourceId,
-		RpcService rpcService,
-		HighAvailabilityServices highAvailabilityServices,
-		HeartbeatServices heartbeatServices,
-		MetricRegistry metricRegistry,
-		FatalErrorHandler fatalErrorHandler) throws Exception {
+			Configuration configuration,
+			ResourceID resourceId,
+			RpcService rpcService,
+			HighAvailabilityServices highAvailabilityServices,
+			HeartbeatServices heartbeatServices,
+			MetricRegistry metricRegistry,
+			FatalErrorHandler fatalErrorHandler,
+			@Nullable String webInterfaceUrl) throws Exception {
 		final ResourceManagerConfiguration rmConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServices rmRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(

flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java

@@ -97,7 +97,8 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
 			highAvailabilityServices,
 			heartbeatServices,
 			metricRegistry,
-			this);
+			this,
+			null);
 
 		jobManagerServices = JobManagerServices.fromConfiguration(configuration, blobServer);
 
@@ -272,7 +273,8 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
 		HighAvailabilityServices highAvailabilityServices,
 		HeartbeatServices heartbeatServices,
 		MetricRegistry metricRegistry,
-		FatalErrorHandler fatalErrorHandler) throws Exception;
+		FatalErrorHandler fatalErrorHandler,
+		@Nullable String webInterfaceUrl) throws Exception;
 
 	protected abstract JobGraph retrieveJobGraph(Configuration configuration) throws FlinkException;
 

flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java

@@ -120,7 +120,8 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
 			highAvailabilityServices,
 			heartbeatServices,
 			metricRegistry,
-			this);
+			this,
+			dispatcherRestEndpoint.getRestAddress());
 
 		dispatcher = createDispatcher(
 			configuration,
@@ -238,5 +239,6 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
 		HighAvailabilityServices highAvailabilityServices,
 		HeartbeatServices heartbeatServices,
 		MetricRegistry metricRegistry,
-		FatalErrorHandler fatalErrorHandler) throws Exception;
+		FatalErrorHandler fatalErrorHandler,
+		@Nullable String webInterfaceUrl) throws Exception;
 }

flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/StandaloneSessionClusterEntrypoint.java

@@ -35,6 +35,8 @@ import org.apache.flink.runtime.util.EnvironmentInformation;
 import org.apache.flink.runtime.util.JvmShutdownSafeguard;
 import org.apache.flink.runtime.util.SignalHandler;
 
+import javax.annotation.Nullable;
+
 /**
  * Entry point for the standalone session cluster.
  */
@@ -52,7 +54,8 @@ public class StandaloneSessionClusterEntrypoint extends SessionClusterEntrypoint
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
 			MetricRegistry metricRegistry,
-			FatalErrorHandler fatalErrorHandler) throws Exception {
+			FatalErrorHandler fatalErrorHandler,
+			@Nullable String webInterfaceUrl) throws Exception {
 		final ResourceManagerConfiguration resourceManagerConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServicesConfiguration resourceManagerRuntimeServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServices resourceManagerRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(

flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java

@@ -52,6 +52,8 @@ import org.apache.hadoop.yarn.client.api.NMClient;
 import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 
+import javax.annotation.Nullable;
+
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -100,6 +102,9 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
 
 	private final YarnConfiguration yarnConfig;
 
+	@Nullable
+	private final String webInterfaceUrl;
+
 	/** Client to communicate with the Resource Manager (YARN's master). */
 	private AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient;
 
@@ -123,7 +128,8 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
 			SlotManager slotManager,
 			MetricRegistry metricRegistry,
 			JobLeaderIdService jobLeaderIdService,
-			FatalErrorHandler fatalErrorHandler) {
+			FatalErrorHandler fatalErrorHandler,
+			@Nullable String webInterfaceUrl) {
 		super(
 			rpcService,
 			resourceManagerEndpointId,
@@ -153,36 +159,50 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
 		}
 		yarnHeartbeatIntervalMillis = yarnHeartbeatIntervalMS;
 		numPendingContainerRequests = 0;
+
+		this.webInterfaceUrl = webInterfaceUrl;
 	}
 
-	protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient() {
-		AMRMClientAsync<AMRMClient.ContainerRequest> rmc = AMRMClientAsync.createAMRMClientAsync(yarnHeartbeatIntervalMillis, this);
-		rmc.init(yarnConfig);
-		rmc.start();
-		try {
-			//TODO: change akka address to tcp host and port, the getAddress() interface should return a standard tcp address
-			Tuple2<String, Integer> hostPort = parseHostPort(getAddress());
-			//TODO: the third paramter should be the webmonitor address
-			rmc.registerApplicationMaster(hostPort.f0, hostPort.f1, getAddress());
-		} catch (Exception e) {
-			log.info("registerApplicationMaster fail", e);
-		}
-		return rmc;
+	protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient(
+			YarnConfiguration yarnConfiguration,
+			int yarnHeartbeatIntervalMillis,
+			@Nullable String webInterfaceUrl) throws Exception {
+		AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient = AMRMClientAsync.createAMRMClientAsync(
+			yarnHeartbeatIntervalMillis,
+			this);
+
+		resourceManagerClient.init(yarnConfiguration);
+		resourceManagerClient.start();
+
+		//TODO: change akka address to tcp host and port, the getAddress() interface should return a standard tcp address
+		Tuple2<String, Integer> hostPort = parseHostPort(getAddress());
+
+		resourceManagerClient.registerApplicationMaster(hostPort.f0, hostPort.f1, webInterfaceUrl);
+
+		return resourceManagerClient;
 	}
 
-	protected NMClient createAndStartNodeManagerClient() {
+	protected NMClient createAndStartNodeManagerClient(YarnConfiguration yarnConfiguration) {
 		// create the client to communicate with the node managers
-		NMClient nmc = NMClient.createNMClient();
-		nmc.init(yarnConfig);
-		nmc.start();
-		nmc.cleanupRunningContainersOnStop(true);
-		return nmc;
+		NMClient nodeManagerClient = NMClient.createNMClient();
+		nodeManagerClient.init(yarnConfiguration);
+		nodeManagerClient.start();
+		nodeManagerClient.cleanupRunningContainersOnStop(true);
+		return nodeManagerClient;
 	}
 
 	@Override
 	protected void initialize() throws ResourceManagerException {
-		resourceManagerClient = createAndStartResourceManagerClient();
-		nodeManagerClient = createAndStartNodeManagerClient();
+		try {
+			resourceManagerClient = createAndStartResourceManagerClient(
+				yarnConfig,
+				yarnHeartbeatIntervalMillis,
+				webInterfaceUrl);
+		} catch (Exception e) {
+			throw new ResourceManagerException("Could not start resource manager client.", e);
+		}
+
+		nodeManagerClient = createAndStartNodeManagerClient(yarnConfig);
 	}
 
 	@Override
@@ -222,7 +242,8 @@ public class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
 
 		// first, de-register from YARN
 		FinalApplicationStatus yarnStatus = getYarnStatus(finalStatus);
-		log.info("Unregistering application from the YARN Resource Manager");
+		log.info("Unregister application from the YARN Resource Manager");
+
 		try {
 			resourceManagerClient.unregisterApplicationMaster(yarnStatus, optionalDiagnostics, "");
 		} catch (Throwable t) {

flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java

@@ -41,6 +41,8 @@ import org.apache.flink.yarn.YarnResourceManager;
 
 import org.apache.hadoop.yarn.api.ApplicationConstants;
 
+import javax.annotation.Nullable;
+
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
@@ -79,7 +81,8 @@ public class YarnJobClusterEntrypoint extends JobClusterEntrypoint {
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
 			MetricRegistry metricRegistry,
-			FatalErrorHandler fatalErrorHandler) throws Exception {
+			FatalErrorHandler fatalErrorHandler,
+			@Nullable String webInterfaceUrl) throws Exception {
 		final ResourceManagerConfiguration rmConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServices rmRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(
@@ -99,7 +102,8 @@ public class YarnJobClusterEntrypoint extends JobClusterEntrypoint {
 			rmRuntimeServices.getSlotManager(),
 			metricRegistry,
 			rmRuntimeServices.getJobLeaderIdService(),
-			fatalErrorHandler);
+			fatalErrorHandler,
+			webInterfaceUrl);
 	}
 
 	@Override

flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnSessionClusterEntrypoint.java

@@ -39,6 +39,8 @@ import org.apache.flink.yarn.YarnResourceManager;
 
 import org.apache.hadoop.yarn.api.ApplicationConstants;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.Map;
 
@@ -69,7 +71,8 @@ public class YarnSessionClusterEntrypoint extends SessionClusterEntrypoint {
 			HighAvailabilityServices highAvailabilityServices,
 			HeartbeatServices heartbeatServices,
 			MetricRegistry metricRegistry,
-			FatalErrorHandler fatalErrorHandler) throws Exception {
+			FatalErrorHandler fatalErrorHandler,
+			@Nullable String webInterfaceUrl) throws Exception {
 		final ResourceManagerConfiguration rmConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration = ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration);
 		final ResourceManagerRuntimeServices rmRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(
@@ -89,7 +92,8 @@ public class YarnSessionClusterEntrypoint extends SessionClusterEntrypoint {
 			rmRuntimeServices.getSlotManager(),
 			metricRegistry,
 			rmRuntimeServices.getJobLeaderIdService(),
-			fatalErrorHandler);
+			fatalErrorHandler,
+			webInterfaceUrl);
 	}
 
 	public static void main(String[] args) {

flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java

@@ -65,6 +65,7 @@ import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.client.api.AMRMClient;
 import org.apache.hadoop.yarn.client.api.NMClient;
 import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -73,6 +74,8 @@ import org.junit.rules.TemporaryFolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.io.File;
 import java.util.HashMap;
 import java.util.Map;
@@ -149,11 +152,23 @@ public class YarnResourceManagerTest extends TestLogger {
 				MetricRegistry metricRegistry,
 				JobLeaderIdService jobLeaderIdService,
 				FatalErrorHandler fatalErrorHandler,
+				@Nullable String webInterfaceUrl,
 				AMRMClientAsync<AMRMClient.ContainerRequest> mockResourceManagerClient,
 				NMClient mockNMClient) {
-			super(rpcService, resourceManagerEndpointId, resourceId, flinkConfig, env,
-					resourceManagerConfiguration, highAvailabilityServices, heartbeatServices,
-					slotManager, metricRegistry, jobLeaderIdService, fatalErrorHandler);
+			super(
+				rpcService,
+				resourceManagerEndpointId,
+				resourceId,
+				flinkConfig,
+				env,
+				resourceManagerConfiguration,
+				highAvailabilityServices,
+				heartbeatServices,
+				slotManager,
+				metricRegistry,
+				jobLeaderIdService,
+				fatalErrorHandler,
+				webInterfaceUrl);
 			this.mockNMClient = mockNMClient;
 			this.mockResourceManagerClient = mockResourceManagerClient;
 		}
@@ -167,12 +182,15 @@ public class YarnResourceManagerTest extends TestLogger {
 		}
 
 		@Override
-		protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient() {
+		protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient(
+				YarnConfiguration yarnConfiguration,
+				int yarnHeartbeatIntervalMillis,
+				@Nullable String webInteraceUrl) {
 			return mockResourceManagerClient;
 		}
 
 		@Override
-		protected NMClient createAndStartNodeManagerClient() {
+		protected NMClient createAndStartNodeManagerClient(YarnConfiguration yarnConfiguration) {
 			return mockNMClient;
 		}
 	}
@@ -231,9 +249,9 @@ public class YarnResourceManagerTest extends TestLogger {
 							rmServices.metricRegistry,
 							rmServices.jobLeaderIdService,
 							fatalErrorHandler,
+							null,
 							mockResourceManagerClient,
-							mockNMClient
-					);
+							mockNMClient);
 		}
 
 		/**