From d5d5a637f6c4f2eeb9430e068c8d620b4010c16d Mon Sep 17 00:00:00 2001
From: rongtong <jinrongtong5@163.com>
Date: Fri, 3 Mar 2023 22:28:15 +0800
Subject: [PATCH] [ISSUE #6226] Shutdown flowMonitor when connection disconnect
 (#6227)

* Shutdown flowMonitor when connection disconnect

* Whether the master is existed has more accurate judgment

* Shutdown flowMonitor when HAClient service end

* Pass the check style

* Modify according to comments
---
 .../rocketmq/controller/impl/manager/SyncStateInfo.java      | 3 ++-
 .../java/org/apache/rocketmq/store/ha/DefaultHAClient.java   | 2 ++
 .../org/apache/rocketmq/store/ha/DefaultHAConnection.java    | 4 ++++
 .../rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java     | 3 +++
 .../rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java | 5 +++++
 5 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/SyncStateInfo.java b/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/SyncStateInfo.java
index 997dee3c56..2674dabf54 100644
--- a/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/SyncStateInfo.java
+++ b/controller/src/main/java/org/apache/rocketmq/controller/impl/manager/SyncStateInfo.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.controller.impl.manager;
 
 import java.util.HashSet;
 import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
 
 /**
  * Manages the syncStateSet of broker replicas.
@@ -53,7 +54,7 @@ public class SyncStateInfo {
     }
 
     public boolean isMasterExist() {
-        return !this.masterAddress.isEmpty();
+        return !StringUtils.isBlank(masterAddress);
     }
 
     public String getClusterName() {
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAClient.java b/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAClient.java
index 06878c185f..530d295ae9 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAClient.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAClient.java
@@ -309,6 +309,7 @@ public class DefaultHAClient extends ServiceThread implements HAClient {
             try {
                 switch (this.currentState) {
                     case SHUTDOWN:
+                        this.flowMonitor.shutdown(true);
                         return;
                     case READY:
                         if (!this.connectMaster()) {
@@ -339,6 +340,7 @@ public class DefaultHAClient extends ServiceThread implements HAClient {
             }
         }
 
+        this.flowMonitor.shutdown(true);
         log.info(this.getServiceName() + " service end");
     }
 
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAConnection.java b/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAConnection.java
index e05e0ce231..5dd24410ed 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAConnection.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/DefaultHAConnection.java
@@ -195,6 +195,8 @@ public class DefaultHAConnection implements HAConnection {
                 log.error("", e);
             }
 
+            flowMonitor.shutdown(true);
+
             log.info(this.getServiceName() + " service end");
         }
 
@@ -398,6 +400,8 @@ public class DefaultHAConnection implements HAConnection {
                 DefaultHAConnection.log.error("", e);
             }
 
+            flowMonitor.shutdown(true);
+
             DefaultHAConnection.log.info(this.getServiceName() + " service end");
         }
 
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
index 554c64fd1a..b95d3814aa 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAClient.java
@@ -398,6 +398,7 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
             try {
                 switch (this.currentState) {
                     case SHUTDOWN:
+                        this.flowMonitor.shutdown(true);
                         return;
                     case READY:
                         // Truncate invalid msg first
@@ -437,6 +438,8 @@ public class AutoSwitchHAClient extends ServiceThread implements HAClient {
             }
         }
 
+        this.flowMonitor.shutdown(true);
+        LOGGER.info(this.getServiceName() + " service end");
     }
 
     /**
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
index 7401574e5e..57f9e9619d 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAConnection.java
@@ -294,6 +294,8 @@ public class AutoSwitchHAConnection implements HAConnection {
                 AutoSwitchHAConnection.LOGGER.error("", e);
             }
 
+            flowMonitor.shutdown(true);
+
             AutoSwitchHAConnection.LOGGER.info(this.getServiceName() + " service end");
         }
 
@@ -739,6 +741,9 @@ public class AutoSwitchHAConnection implements HAConnection {
             } catch (IOException e) {
                 AutoSwitchHAConnection.LOGGER.error("", e);
             }
+
+            flowMonitor.shutdown(true);
+
             AutoSwitchHAConnection.LOGGER.info(this.getServiceName() + " service end");
         }