You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/24 11:32:48 UTC
[09/10] flink git commit: [FLINK-8677] [flip6] Make ClusterEntrypoint
shut down non-blocking
[FLINK-8677] [flip6] Make ClusterEntrypoint shut down non-blocking
Makes the ClusterEntrypoint shut down method non-blocking. This also removes
the need to run the Dispatcher#terminationFuture callback in the common
Fork-Join pool.
This closes #5512.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fba655af
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fba655af
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fba655af
Branch: refs/heads/master
Commit: fba655af1a4a543f01eecf222826a0821e6e1c17
Parents: 65081ac
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Feb 16 10:08:25 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Fri Feb 23 18:22:10 2018 +0100
----------------------------------------------------------------------
.../entrypoint/MesosJobClusterEntrypoint.java | 33 ++-
.../MesosSessionClusterEntrypoint.java | 34 ++--
.../runtime/entrypoint/ClusterEntrypoint.java | 202 ++++++++++---------
3 files changed, 127 insertions(+), 142 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/fba655af/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
index b0054bb..02f561a 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
@@ -28,6 +28,7 @@ import org.apache.flink.mesos.util.MesosConfiguration;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContainerSpecification;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.entrypoint.JobClusterEntrypoint;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
@@ -43,7 +44,6 @@ import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.runtime.util.JvmShutdownSafeguard;
import org.apache.flink.runtime.util.SignalHandler;
-import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
@@ -59,6 +59,7 @@ import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.ObjectInputStream;
+import java.util.concurrent.CompletableFuture;
/**
* Entry point for Mesos per-job clusters.
@@ -167,26 +168,16 @@ public class MesosJobClusterEntrypoint extends JobClusterEntrypoint {
}
@Override
- protected void stopClusterServices(boolean cleanupHaData) throws FlinkException {
- Throwable exception = null;
-
- try {
- super.stopClusterServices(cleanupHaData);
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
-
- if (mesosServices != null) {
- try {
- mesosServices.close(cleanupHaData);
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (exception != null) {
- throw new FlinkException("Could not properly shut down the Mesos job cluster entry point.", exception);
- }
+ protected CompletableFuture<Void> stopClusterServices(boolean cleanupHaData) {
+ final CompletableFuture<Void> serviceShutDownFuture = super.stopClusterServices(cleanupHaData);
+
+ return FutureUtils.runAfterwards(
+ serviceShutDownFuture,
+ () -> {
+ if (mesosServices != null) {
+ mesosServices.close(cleanupHaData);
+ }
+ });
}
public static void main(String[] args) {
http://git-wip-us.apache.org/repos/asf/flink/blob/fba655af/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
index 306d5b2..5dea936 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
@@ -28,6 +28,7 @@ import org.apache.flink.mesos.util.MesosConfiguration;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContainerSpecification;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.entrypoint.SessionClusterEntrypoint;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
@@ -42,7 +43,6 @@ import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.runtime.util.JvmShutdownSafeguard;
import org.apache.flink.runtime.util.SignalHandler;
-import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.commons.cli.CommandLine;
@@ -52,6 +52,8 @@ import org.apache.commons.cli.PosixParser;
import javax.annotation.Nullable;
+import java.util.concurrent.CompletableFuture;
+
/**
* Entry point for Mesos session clusters.
*/
@@ -141,26 +143,16 @@ public class MesosSessionClusterEntrypoint extends SessionClusterEntrypoint {
}
@Override
- protected void stopClusterServices(boolean cleanupHaData) throws FlinkException {
- Throwable exception = null;
-
- try {
- super.stopClusterServices(cleanupHaData);
- } catch (Throwable t) {
- exception = t;
- }
-
- if (mesosServices != null) {
- try {
- mesosServices.close(cleanupHaData);
- } catch (Throwable t) {
- exception = t;
- }
- }
-
- if (exception != null) {
- throw new FlinkException("Could not properly shut down the Mesos session cluster entry point.", exception);
- }
+ protected CompletableFuture<Void> stopClusterServices(boolean cleanupHaData) {
+ final CompletableFuture<Void> serviceShutDownFuture = super.stopClusterServices(cleanupHaData);
+
+ return FutureUtils.runAfterwards(
+ serviceShutDownFuture,
+ () -> {
+ if (mesosServices != null) {
+ mesosServices.close(cleanupHaData);
+ }
+ });
}
public static void main(String[] args) {
http://git-wip-us.apache.org/repos/asf/flink/blob/fba655af/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index 8e4f936..ba28670 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.blob.TransientBlobService;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
import org.apache.flink.runtime.dispatcher.ArchivedExecutionGraphStore;
import org.apache.flink.runtime.dispatcher.Dispatcher;
@@ -63,7 +64,6 @@ import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever
import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
import org.apache.flink.runtime.webmonitor.retriever.impl.RpcGatewayRetriever;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import akka.actor.ActorSystem;
@@ -75,6 +75,8 @@ import javax.annotation.concurrent.GuardedBy;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
@@ -104,10 +106,12 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
private final Configuration configuration;
- private final CompletableFuture<Boolean> terminationFuture;
+ private final CompletableFuture<Void> terminationFuture;
private final AtomicBoolean isTerminating = new AtomicBoolean(false);
+ private final AtomicBoolean isShutDown = new AtomicBoolean(false);
+
@GuardedBy("lock")
private MetricRegistryImpl metricRegistry;
@@ -144,12 +148,15 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
@GuardedBy("lock")
private TransientBlobCache transientBlobCache;
+ @GuardedBy("lock")
+ private ClusterInformation clusterInformation;
+
protected ClusterEntrypoint(Configuration configuration) {
this.configuration = Preconditions.checkNotNull(configuration);
this.terminationFuture = new CompletableFuture<>();
}
- public CompletableFuture<Boolean> getTerminationFuture() {
+ public CompletableFuture<Void> getTerminationFuture() {
return terminationFuture;
}
@@ -211,8 +218,7 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
heartbeatServices,
metricRegistry);
- // TODO: Make shutDownAndTerminate non blocking to not use the global executor
- dispatcher.getTerminationFuture().whenCompleteAsync(
+ dispatcher.getTerminationFuture().whenComplete(
(Void value, Throwable throwable) -> {
if (throwable != null) {
LOG.info("Could not properly terminate the Dispatcher.", throwable);
@@ -250,6 +256,18 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
// start the MetricQueryService
final ActorSystem actorSystem = ((AkkaRpcService) commonRpcService).getActorSystem();
metricRegistry.startQueryService(actorSystem, null);
+
+ archivedExecutionGraphStore = createSerializableExecutionGraphStore(configuration, commonRpcService.getScheduledExecutor());
+
+ clusterInformation = new ClusterInformation(
+ commonRpcService.getAddress(),
+ blobServer.getPort());
+
+ transientBlobCache = new TransientBlobCache(
+ configuration,
+ new InetSocketAddress(
+ clusterInformation.getBlobServerHostname(),
+ clusterInformation.getBlobServerPort()));
}
}
@@ -261,22 +279,10 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry) throws Exception {
synchronized (lock) {
- archivedExecutionGraphStore = createSerializableExecutionGraphStore(configuration, rpcService.getScheduledExecutor());
-
dispatcherLeaderRetrievalService = highAvailabilityServices.getDispatcherLeaderRetriever();
resourceManagerRetrievalService = highAvailabilityServices.getResourceManagerLeaderRetriever();
- final ClusterInformation clusterInformation = new ClusterInformation(
- rpcService.getAddress(),
- blobServer.getPort());
-
- transientBlobCache = new TransientBlobCache(
- configuration,
- new InetSocketAddress(
- clusterInformation.getBlobServerHostname(),
- clusterInformation.getBlobServerPort()));
-
LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever = new RpcGatewayRetriever<>(
rpcService,
DispatcherGateway.class,
@@ -376,44 +382,11 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
return new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(configuration));
}
- private void shutDown(boolean cleanupHaData) throws FlinkException {
- LOG.info("Stopping {}.", getClass().getSimpleName());
-
- Throwable exception = null;
-
+ protected CompletableFuture<Void> stopClusterServices(boolean cleanupHaData) {
synchronized (lock) {
+ Throwable exception = null;
- try {
- stopClusterComponents();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
-
- try {
- stopClusterServices(cleanupHaData);
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
-
- terminationFuture.complete(true);
- }
-
- if (exception != null) {
- throw new FlinkException("Could not properly shut down the cluster entrypoint.", exception);
- }
- }
-
- protected void stopClusterServices(boolean cleanupHaData) throws FlinkException {
- Throwable exception = null;
-
- synchronized (lock) {
- if (metricRegistry != null) {
- try {
- metricRegistry.shutdown();
- } catch (Throwable t) {
- exception = t;
- }
- }
+ final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3);
if (blobServer != null) {
try {
@@ -435,79 +408,80 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
}
}
- if (commonRpcService != null) {
+ if (archivedExecutionGraphStore != null) {
try {
- commonRpcService.stopService().get();
+ archivedExecutionGraphStore.close();
} catch (Throwable t) {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}
}
- }
- if (exception != null) {
- throw new FlinkException("Could not properly shut down the cluster services.", exception);
+ if (transientBlobCache != null) {
+ try {
+ transientBlobCache.close();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+ }
+
+ if (metricRegistry != null) {
+ terminationFutures.add(metricRegistry.shutdown());
+ }
+
+ if (commonRpcService != null) {
+ terminationFutures.add(commonRpcService.stopService());
+ }
+
+ if (exception != null) {
+ terminationFutures.add(FutureUtils.completedExceptionally(exception));
+ }
+
+ return FutureUtils.completeAll(terminationFutures);
}
}
- protected void stopClusterComponents() throws Exception {
+ protected CompletableFuture<Void> stopClusterComponents() {
synchronized (lock) {
- Throwable exception = null;
- if (webMonitorEndpoint != null) {
- webMonitorEndpoint.shutDownAsync().get();
- }
+ Exception exception = null;
+
+ final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(4);
if (dispatcherLeaderRetrievalService != null) {
try {
dispatcherLeaderRetrievalService.stop();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (dispatcher != null) {
- try {
- dispatcher.shutDown();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
}
}
if (resourceManagerRetrievalService != null) {
try {
resourceManagerRetrievalService.stop();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
}
}
- if (resourceManager != null) {
- try {
- resourceManager.shutDown();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
+ if (webMonitorEndpoint != null) {
+ terminationFutures.add(webMonitorEndpoint.shutDownAsync());
}
- if (archivedExecutionGraphStore != null) {
- try {
- archivedExecutionGraphStore.close();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
+ if (dispatcher != null) {
+ dispatcher.shutDown();
+ terminationFutures.add(dispatcher.getTerminationFuture());
}
- if (transientBlobCache != null) {
- try {
- transientBlobCache.close();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
+ if (resourceManager != null) {
+ resourceManager.shutDown();
+ terminationFutures.add(resourceManager.getTerminationFuture());
}
if (exception != null) {
- throw new FlinkException("Could not properly shut down the session cluster entry point.", exception);
+ terminationFutures.add(FutureUtils.completedExceptionally(exception));
}
+
+ return FutureUtils.completeAll(terminationFutures);
}
}
@@ -522,6 +496,33 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
// Internal methods
// --------------------------------------------------
+ private CompletableFuture<Void> shutDownAsync(boolean cleanupHaData) {
+ if (isShutDown.compareAndSet(false, true)) {
+ LOG.info("Stopping {}.", getClass().getSimpleName());
+
+ final CompletableFuture<Void> componentShutdownFuture = stopClusterComponents();
+
+ componentShutdownFuture.whenComplete(
+ (Void ignored1, Throwable componentThrowable) -> {
+ final CompletableFuture<Void> serviceShutdownFuture = stopClusterServices(cleanupHaData);
+
+ serviceShutdownFuture.whenComplete(
+ (Void ignored2, Throwable serviceThrowable) -> {
+ if (serviceThrowable != null) {
+ terminationFuture.completeExceptionally(
+ ExceptionUtils.firstOrSuppressed(serviceThrowable, componentThrowable));
+ } else if (componentThrowable != null) {
+ terminationFuture.completeExceptionally(componentThrowable);
+ } else {
+ terminationFuture.complete(null);
+ }
+ });
+ });
+ }
+
+ return terminationFuture;
+ }
+
private void shutDownAndTerminate(
int returnCode,
ApplicationStatus applicationStatus,
@@ -533,13 +534,14 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
applicationStatus);
if (isTerminating.compareAndSet(false, true)) {
- try {
- shutDown(cleanupHaData);
- } catch (Throwable t) {
- LOG.info("Could not properly shut down cluster entrypoint.", t);
- }
+ shutDownAsync(cleanupHaData).whenComplete(
+ (Void ignored, Throwable t) -> {
+ if (t != null) {
+ LOG.info("Could not properly shut down cluster entrypoint.", t);
+ }
- System.exit(returnCode);
+ System.exit(returnCode);
+ });
} else {
LOG.debug("Concurrent termination call detected. Ignoring termination call with return code {} and application status {}.",
returnCode,