[FLINK-14305] Transfer ownership of JobManagerMetricGroup to Dispatcher

With this commit the Dispatcher is now responsible for managing the lifecycle of the JobManagerMetricGroup.
apache · Oct 2, 2019 · 5c23b74 · 5c23b74
1 parent d886509
commit 5c23b74
Show file tree

Hide file tree

Showing 6 changed files with 12 additions and 34 deletions.
diff --git a/...flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java b/...flink/runtime/entrypoint/component/AbstractDispatcherResourceManagerComponentFactory.java
@@ -212,8 +212,7 @@ public DispatcherResourceManagerComponent<T> create(
 				resourceManager,
 				dispatcherLeaderRetrievalService,
 				resourceManagerRetrievalService,
-				webMonitorEndpoint,
-				jobManagerMetricGroup);
+				webMonitorEndpoint);
 
 		} catch (Exception exception) {
 			// clean up all started components
@@ -272,6 +271,5 @@ protected abstract DispatcherResourceManagerComponent<T> createDispatcherResourc
 		ResourceManager<?> resourceManager,
 		LeaderRetrievalService dispatcherLeaderRetrievalService,
 		LeaderRetrievalService resourceManagerRetrievalService,
-		WebMonitorEndpoint<?> webMonitorEndpoint,
-		JobManagerMetricGroup jobManagerMetricGroup);
+		WebMonitorEndpoint<?> webMonitorEndpoint);
 }
diff --git a/...ava/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java b/...ava/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java
@@ -22,7 +22,6 @@
 import org.apache.flink.runtime.concurrent.FutureUtils;
 import org.apache.flink.runtime.dispatcher.Dispatcher;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManager;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
@@ -58,9 +57,6 @@ public class DispatcherResourceManagerComponent<T extends Dispatcher> implements
 	@Nonnull
 	private final WebMonitorEndpoint<?> webMonitorEndpoint;
 
-	@Nonnull
-	private final JobManagerMetricGroup jobManagerMetricGroup;
-
 	private final CompletableFuture<Void> terminationFuture;
 
 	private final CompletableFuture<ApplicationStatus> shutDownFuture;
@@ -72,14 +68,12 @@ public class DispatcherResourceManagerComponent<T extends Dispatcher> implements
 			@Nonnull ResourceManager<?> resourceManager,
 			@Nonnull LeaderRetrievalService dispatcherLeaderRetrievalService,
 			@Nonnull LeaderRetrievalService resourceManagerRetrievalService,
-			@Nonnull WebMonitorEndpoint<?> webMonitorEndpoint,
-			@Nonnull JobManagerMetricGroup jobManagerMetricGroup) {
+			@Nonnull WebMonitorEndpoint<?> webMonitorEndpoint) {
 		this.resourceManager = resourceManager;
 		this.dispatcher = dispatcher;
 		this.dispatcherLeaderRetrievalService = dispatcherLeaderRetrievalService;
 		this.resourceManagerRetrievalService = resourceManagerRetrievalService;
 		this.webMonitorEndpoint = webMonitorEndpoint;
-		this.jobManagerMetricGroup = jobManagerMetricGroup;
 		this.terminationFuture = new CompletableFuture<>();
 		this.shutDownFuture = new CompletableFuture<>();
 
@@ -179,11 +173,7 @@ private CompletableFuture<Void> closeAsyncInternal() {
 
 		final CompletableFuture<Void> componentTerminationFuture = FutureUtils.completeAll(terminationFutures);
 
-		final CompletableFuture<Void> metricGroupTerminationFuture = FutureUtils.runAfterwards(
-			componentTerminationFuture,
-			jobManagerMetricGroup::close);
-
-		metricGroupTerminationFuture.whenComplete((aVoid, throwable) -> {
+		componentTerminationFuture.whenComplete((aVoid, throwable) -> {
 			if (throwable != null) {
 				terminationFuture.completeExceptionally(throwable);
 			} else {

diff --git a/.../org/apache/flink/runtime/entrypoint/component/JobDispatcherResourceManagerComponent.java b/.../org/apache/flink/runtime/entrypoint/component/JobDispatcherResourceManagerComponent.java
@@ -21,7 +21,6 @@
 import org.apache.flink.runtime.clusterframework.ApplicationStatus;
 import org.apache.flink.runtime.dispatcher.MiniDispatcher;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManager;
 import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
 
@@ -38,9 +37,8 @@ class JobDispatcherResourceManagerComponent extends DispatcherResourceManagerCom
 			ResourceManager<?> resourceManager,
 			LeaderRetrievalService dispatcherLeaderRetrievalService,
 			LeaderRetrievalService resourceManagerRetrievalService,
-			WebMonitorEndpoint<?> webMonitorEndpoint,
-			JobManagerMetricGroup jobManagerMetricGroup) {
-		super(dispatcher, resourceManager, dispatcherLeaderRetrievalService, resourceManagerRetrievalService, webMonitorEndpoint, jobManagerMetricGroup);
+			WebMonitorEndpoint<?> webMonitorEndpoint) {
+		super(dispatcher, resourceManager, dispatcherLeaderRetrievalService, resourceManagerRetrievalService, webMonitorEndpoint);
 
 		final CompletableFuture<ApplicationStatus> shutDownFuture = getShutDownFuture();
 

diff --git a/...ache/flink/runtime/entrypoint/component/JobDispatcherResourceManagerComponentFactory.java b/...ache/flink/runtime/entrypoint/component/JobDispatcherResourceManagerComponentFactory.java
@@ -21,7 +21,6 @@
 import org.apache.flink.runtime.dispatcher.JobDispatcherFactory;
 import org.apache.flink.runtime.dispatcher.MiniDispatcher;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManager;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory;
 import org.apache.flink.runtime.rest.JobRestEndpointFactory;
@@ -45,14 +44,12 @@ protected DispatcherResourceManagerComponent<MiniDispatcher> createDispatcherRes
 			ResourceManager<?> resourceManager,
 			LeaderRetrievalService dispatcherLeaderRetrievalService,
 			LeaderRetrievalService resourceManagerRetrievalService,
-			WebMonitorEndpoint<?> webMonitorEndpoint,
-			JobManagerMetricGroup jobManagerMetricGroup) {
+			WebMonitorEndpoint<?> webMonitorEndpoint) {
 		return new JobDispatcherResourceManagerComponent(
 			dispatcher,
 			resourceManager,
 			dispatcherLeaderRetrievalService,
 			resourceManagerRetrievalService,
-			webMonitorEndpoint,
-			jobManagerMetricGroup);
+			webMonitorEndpoint);
 	}
 }
diff --git a/.../apache/flink/runtime/entrypoint/component/SessionDispatcherResourceManagerComponent.java b/.../apache/flink/runtime/entrypoint/component/SessionDispatcherResourceManagerComponent.java
@@ -20,7 +20,6 @@
 
 import org.apache.flink.runtime.dispatcher.Dispatcher;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManager;
 import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
 
@@ -33,8 +32,7 @@ class SessionDispatcherResourceManagerComponent extends DispatcherResourceManage
 			ResourceManager<?> resourceManager,
 			LeaderRetrievalService dispatcherLeaderRetrievalService,
 			LeaderRetrievalService resourceManagerRetrievalService,
-			WebMonitorEndpoint<?> webMonitorEndpoint,
-			JobManagerMetricGroup jobManagerMetricGroup) {
-		super(dispatcher, resourceManager, dispatcherLeaderRetrievalService, resourceManagerRetrievalService, webMonitorEndpoint, jobManagerMetricGroup);
+			WebMonitorEndpoint<?> webMonitorEndpoint) {
+		super(dispatcher, resourceManager, dispatcherLeaderRetrievalService, resourceManagerRetrievalService, webMonitorEndpoint);
 	}
 }
diff --git a/.../flink/runtime/entrypoint/component/SessionDispatcherResourceManagerComponentFactory.java b/.../flink/runtime/entrypoint/component/SessionDispatcherResourceManagerComponentFactory.java
@@ -24,7 +24,6 @@
 import org.apache.flink.runtime.dispatcher.DispatcherGateway;
 import org.apache.flink.runtime.dispatcher.SessionDispatcherFactory;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManager;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory;
 import org.apache.flink.runtime.rest.SessionRestEndpointFactory;
@@ -54,14 +53,12 @@ protected DispatcherResourceManagerComponent<Dispatcher> createDispatcherResourc
 			ResourceManager<?> resourceManager,
 			LeaderRetrievalService dispatcherLeaderRetrievalService,
 			LeaderRetrievalService resourceManagerRetrievalService,
-			WebMonitorEndpoint<?> webMonitorEndpoint,
-			JobManagerMetricGroup jobManagerMetricGroup) {
+			WebMonitorEndpoint<?> webMonitorEndpoint) {
 		return new SessionDispatcherResourceManagerComponent(
 			dispatcher,
 			resourceManager,
 			dispatcherLeaderRetrievalService,
 			resourceManagerRetrievalService,
-			webMonitorEndpoint,
-			jobManagerMetricGroup);
+			webMonitorEndpoint);
 	}
 }