(flink) branch master updated: [FLINK-36194][Runtime] Handle the Graceful close of ExecutionGraphInfoStore from ClusterEntrypoint

[mapohl]

This is an automated email from the ASF dual-hosted git repository.

mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 82ca1042870 [FLINK-36194][Runtime] Handle the Graceful close of 
ExecutionGraphInfoStore from ClusterEntrypoint
82ca1042870 is described below

commit 82ca1042870a5e9b44da8b3a76ea6c9e20f39c7a
Author: Eaugene Thomas <teaug...@uber.com>
AuthorDate: Sat Nov 30 13:24:52 2024 +0530

    [FLINK-36194][Runtime] Handle the Graceful close of ExecutionGraphInfoStore 
from ClusterEntrypoint
---
 .../flink/runtime/dispatcher/FileExecutionGraphInfoStore.java     | 8 --------
 .../flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java   | 7 -------
 .../org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java    | 1 +
 3 files changed, 1 insertion(+), 15 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStore.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStore.java
index f3863adf1a5..470678f7657 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStore.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStore.java
@@ -28,7 +28,6 @@ import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
 import org.apache.flink.util.FileUtils;
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.ShutdownHookUtil;
 import org.apache.flink.util.concurrent.ScheduledExecutor;
 
 import org.apache.flink.shaded.guava32.com.google.common.base.Ticker;
@@ -72,8 +71,6 @@ public class FileExecutionGraphInfoStore implements 
ExecutionGraphInfoStore {
 
     private final ScheduledFuture<?> cleanupFuture;
 
-    private final Thread shutdownHook;
-
     private int numFinishedJobs;
 
     private int numFailedJobs;
@@ -132,8 +129,6 @@ public class FileExecutionGraphInfoStore implements 
ExecutionGraphInfoStore {
                         expirationTime.toMillis(),
                         TimeUnit.MILLISECONDS);
 
-        this.shutdownHook = ShutdownHookUtil.addShutdownHook(this, 
getClass().getSimpleName(), LOG);
-
         this.numFinishedJobs = 0;
         this.numFailedJobs = 0;
         this.numCanceledJobs = 0;
@@ -232,9 +227,6 @@ public class FileExecutionGraphInfoStore implements 
ExecutionGraphInfoStore {
 
         // clean up the storage directory
         FileUtils.deleteFileOrDirectory(storageDir);
-
-        // Remove shutdown hook to prevent resource leaks
-        ShutdownHookUtil.removeShutdownHook(shutdownHook, 
getClass().getSimpleName(), LOG);
     }
 
     // --------------------------------------------------------------
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java
index f8d1daeb5c0..b9ee4c13548 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java
@@ -24,7 +24,6 @@ import 
org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
 import org.apache.flink.runtime.messages.webmonitor.JobDetails;
 import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
 import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
-import org.apache.flink.util.ShutdownHookUtil;
 import org.apache.flink.util.concurrent.ScheduledExecutor;
 
 import org.apache.flink.shaded.guava32.com.google.common.base.Ticker;
@@ -55,8 +54,6 @@ public class MemoryExecutionGraphInfoStore implements 
ExecutionGraphInfoStore {
 
     @Nullable private final ScheduledFuture<?> cleanupFuture;
 
-    private final Thread shutdownHook;
-
     public MemoryExecutionGraphInfoStore() {
         this(Duration.ofMillis(0), 0, null, null);
     }
@@ -89,7 +86,6 @@ public class MemoryExecutionGraphInfoStore implements 
ExecutionGraphInfoStore {
         } else {
             this.cleanupFuture = null;
         }
-        this.shutdownHook = ShutdownHookUtil.addShutdownHook(this, 
getClass().getSimpleName(), LOG);
     }
 
     @Override
@@ -149,8 +145,5 @@ public class MemoryExecutionGraphInfoStore implements 
ExecutionGraphInfoStore {
         }
 
         serializableExecutionGraphInfos.invalidateAll();
-
-        // Remove shutdown hook to prevent resource leaks
-        ShutdownHookUtil.removeShutdownHook(shutdownHook, 
getClass().getSimpleName(), LOG);
     }
 }
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 f9b99e105e0..122db8d639f 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
@@ -510,6 +510,7 @@ public abstract class ClusterEntrypoint implements 
AutoCloseableAsync, FatalErro
             if (executionGraphInfoStore != null) {
                 try {
                     executionGraphInfoStore.close();
+                    executionGraphInfoStore = null;
                 } catch (Throwable t) {
                     exception = ExceptionUtils.firstOrSuppressed(t, exception);
                 }