(gobblin) branch master updated: [GOBBLIN-2038] Fix bug where monitor startup is never occurring (#3917)

[wlo]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new a25c8399b [GOBBLIN-2038] Fix bug where monitor startup is never 
occurring (#3917)
a25c8399b is described below

commit a25c8399b00e01834d8af6b2e99d6bbd64386bc5
Author: William Lo <lo.willia...@gmail.com>
AuthorDate: Mon Apr 8 18:34:54 2024 -0400

    [GOBBLIN-2038] Fix bug where monitor startup is never occurring (#3917)
---
 .../monitoring/DagActionStoreChangeMonitor.java    | 22 ++++++++++------------
 .../service/monitoring/SpecStoreChangeMonitor.java |  6 ++----
 2 files changed, 12 insertions(+), 16 deletions(-)

diff --git 
a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
 
b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
index c9258c013..a3abb6454 100644
--- 
a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
+++ 
b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/DagActionStoreChangeMonitor.java
@@ -165,18 +165,16 @@ public class DagActionStoreChangeMonitor extends 
HighLevelConsumer {
       return;
     }
 
-    if (isActive) {
-      this.isActive = true;
-      initializeMonitor();
-      // Method that starts threads that processes queues
-      processQueues();
-      // Main thread that constantly polls messages from kafka
-      consumerExecutor.execute(() -> {
-        while (!shutdownRequested) {
-          consume();
-        }
-      });
-    }
+    this.isActive = true;
+    initializeMonitor();
+    // Method that starts threads that processes queues
+    processQueues();
+    // Main thread that constantly polls messages from kafka
+    consumerExecutor.execute(() -> {
+      while (!shutdownRequested) {
+        consume();
+      }
+    });
   }
 
   @Override
diff --git 
a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/SpecStoreChangeMonitor.java
 
b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/SpecStoreChangeMonitor.java
index 86cb911d1..03817caca 100644
--- 
a/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/SpecStoreChangeMonitor.java
+++ 
b/gobblin-service/src/main/java/org/apache/gobblin/service/monitoring/SpecStoreChangeMonitor.java
@@ -114,10 +114,8 @@ public class SpecStoreChangeMonitor extends 
HighLevelConsumer {
       return;
     }
 
-    if (isActive) {
-      this.isActive = true;
-      super.startUp();
-    }
+    this.isActive = true;
+    super.startUp();
   }
 
   @Override