[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

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() {

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");
     }
 

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");
         }
 

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");
     }
 
     /**

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");
         }