[hotfix][runtime] Adds shutdown logic for the watchExecutorService to…

… the k8s HA services Signed-off-by: Matthias Pohl <[email protected]>
jonchase · Jul 6, 2023 · 08d0e6b · 08d0e6b
1 parent a46cf26
commit 08d0e6b
Showing 1 changed file with 12 additions and 0 deletions.
diff --git a/...link/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServices.java b/...link/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServices.java
@@ -37,6 +37,9 @@
 import org.apache.flink.util.ExecutorUtils;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.UUID;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
@@ -50,6 +53,9 @@
 /** Kubernetes HA services that use a single leader election service per JobManager. */
 public class KubernetesMultipleComponentLeaderElectionHaServices extends AbstractHaServices {
 
+    private static final Logger LOG =
+            LoggerFactory.getLogger(KubernetesMultipleComponentLeaderElectionHaServices.class);
+
     private final String clusterId;
 
     private final FlinkKubeClient kubeClient;
@@ -178,6 +184,12 @@ public void internalClose() throws Exception {
 
     private void closeK8sServices() {
         configMapSharedWatcher.close();
+        final int outstandingTaskCount = watchExecutorService.shutdownNow().size();
+        if (outstandingTaskCount != 0) {
+            LOG.debug(
+                    "The k8s HA services were closed with {} event(s) still not being processed. No further action necessary.",
+                    outstandingTaskCount);
+        }
     }
 
     @Override