[hotfix] Let ClusterEntrypoint implement AutoCloseableAsync

tongcheng-elong · Sep 28, 2018 · 20447b2 · 20447b2
1 parent 87c27e4
commit 20447b2
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 8 deletions.
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
@@ -54,6 +54,7 @@
 import org.apache.flink.runtime.security.SecurityContext;
 import org.apache.flink.runtime.security.SecurityUtils;
 import org.apache.flink.runtime.util.ZooKeeperUtils;
+import org.apache.flink.util.AutoCloseableAsync;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.FileUtils;
 import org.apache.flink.util.Preconditions;
@@ -88,7 +89,7 @@
  *
  * <p>Specialization of this class can be used for the session mode and the per-job mode
  */
-public abstract class ClusterEntrypoint implements FatalErrorHandler {
+public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErrorHandler {
 
  public static final ConfigOption<String> EXECUTION_MODE = ConfigOptions
  .key("internal.cluster.execution-mode")
@@ -312,6 +313,14 @@ protected MetricRegistryImpl createMetricRegistry(Configuration configuration) {
  return new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(configuration));
  }
 
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return shutDownAsync(
+ ApplicationStatus.UNKNOWN,
+ "Cluster entrypoint has been closed externally.",
+ true).thenAccept(ignored -> {});
+ }
+
  protected CompletableFuture<Void> stopClusterServices(boolean cleanupHaData) {
  synchronized (lock) {
  Throwable exception = null;
@@ -392,7 +401,7 @@ private Configuration generateClusterConfiguration(Configuration configuration)
  return resultConfiguration;
  }
 
- public CompletableFuture<ApplicationStatus> shutDownAsync(
+ private CompletableFuture<ApplicationStatus> shutDownAsync(
  ApplicationStatus applicationStatus,
  @Nullable String diagnostics,
  boolean cleanupHaData) {

diff --git a/...st/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java b/...st/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java
@@ -24,7 +24,6 @@
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.runtime.clusterframework.ApplicationStatus;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint;
 import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
@@ -101,9 +100,7 @@ public void testTaskManagerProcessFailure() throws Exception {
  jmConfig.setLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT, 10000L);
  jmConfig.setInteger(RestOptions.PORT, restPort);
 
- final StandaloneSessionClusterEntrypoint clusterEntrypoint = new StandaloneSessionClusterEntrypoint(jmConfig);
-
- try {
+ try (final StandaloneSessionClusterEntrypoint clusterEntrypoint = new StandaloneSessionClusterEntrypoint(jmConfig)) {
  // check that we run this test only if the java command
  // is available on this machine
  String javaCommand = getJavaCommandPath();
@@ -228,8 +225,6 @@ public void run() {
  if (taskManagerProcess3 != null) {
  taskManagerProcess3.destroy();
  }
-
- clusterEntrypoint.shutDownAsync(ApplicationStatus.SUCCEEDED, null, true).get();
  }
  }