From 40ad03f123f8a170ec6b5172699fb4b8f801b11c Mon Sep 17 00:00:00 2001
From: "github-actions[bot]" <github-actions[bot]@users.noreply.github.com>
Date: Wed, 20 Mar 2024 04:49:21 +0000
Subject: [PATCH] FlintStreamingJobCleanerTask Implementation (#2574)

Signed-off-by: Vamsi Manohar <reddyvam@amazon.com>
(cherry picked from commit 284a0beca7012c1f57b8160c46ab03ed781981f9)
Signed-off-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() {