From faa1d6207fb60e4393c27b48e3cfdcf82154377c Mon Sep 17 00:00:00 2001
From: "opensearch-trigger-bot[bot]"
 <98922864+opensearch-trigger-bot[bot]@users.noreply.github.com>
Date: Wed, 20 Mar 2024 07:23:59 -0700
Subject: [PATCH] FlintStreamingJobCleanerTask Implementation (#2574) (#2576)

(cherry picked from commit 284a0beca7012c1f57b8160c46ab03ed781981f9)

Signed-off-by: Vamsi Manohar <reddyvam@amazon.com>
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
---
 .../spark/cluster/ClusterManagerEventListener.java  | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/spark/src/main/java/org/opensearch/sql/spark/cluster/ClusterManagerEventListener.java b/spark/src/main/java/org/opensearch/sql/spark/cluster/ClusterManagerEventListener.java
index 8f38583b3f..f04c6cb830 100644
--- a/spark/src/main/java/org/opensearch/sql/spark/cluster/ClusterManagerEventListener.java
+++ b/spark/src/main/java/org/opensearch/sql/spark/cluster/ClusterManagerEventListener.java
@@ -133,7 +133,18 @@ public void beforeStop() {
             }
           });
     }
-    initializeStreamingJobHouseKeeperCron();
+
+    if (flintStreamingJobHouseKeeperCron == null) {
+      initializeStreamingJobHouseKeeperCron();
+      clusterService.addLifecycleListener(
+          new LifecycleListener() {
+            @Override
+            public void beforeStop() {
+              cancel(flintStreamingJobHouseKeeperCron);
+              flintStreamingJobHouseKeeperCron = null;
+            }
+          });
+    }
   }
 
   private void initializeStreamingJobHouseKeeperCron() {