[BugFix] When the status of HeartBeat is not ok, it also need to be synced to follower

fe/fe-core/src/main/java/com/starrocks/catalog/FsBroker.java

@@ -82,10 +82,15 @@ public boolean handleHbResponse(BrokerHbResponse hbResponse) {
             } else {
                 if (isAlive) {
                     isAlive = false;
-                    isChanged = true;
                 }
                 heartbeatErrMsg = hbResponse.getMsg() == null ? "Unknown error" : hbResponse.getMsg();
             }
+            // When the master receives an error heartbeat info which status not ok, 
+            // this heartbeat info also need to be synced to follower.
+            // Since the failed heartbeat info also modifies fe's memory, (this.heartbeatRetryTimes++;)
+            // if this heartbeat is not synchronized to the follower, 
+            // that will cause the Follower and master’s memory to be inconsistent
+            isChanged = true;
         }
 
         return isChanged;

fe/fe-core/src/main/java/com/starrocks/system/ComputeNode.java

@@ -401,13 +401,17 @@ public boolean handleHbResponse(BackendHbResponse hbResponse) {
                 this.heartbeatRetryTimes++;
             } else {
                 if (isAlive.compareAndSet(true, false)) {
-                    isChanged = true;
                     LOG.info("{} is dead,", this.toString());
                 }
-
                 heartbeatErrMsg = hbResponse.getMsg() == null ? "Unknown error" : hbResponse.getMsg();
                 lastMissingHeartbeatTime = System.currentTimeMillis();
             }
+            // When the master receives an error heartbeat info which status not ok, 
+            // this heartbeat info also need to be synced to follower.
+            // Since the failed heartbeat info also modifies fe's memory, (this.heartbeatRetryTimes++;)
+            // if this heartbeat is not synchronized to the follower, 
+            // that will cause the Follower and master’s memory to be inconsistent
+            isChanged = true;
         }
 
         return isChanged;

fe/fe-core/src/main/java/com/starrocks/system/Frontend.java

@@ -147,10 +147,15 @@ public boolean handleHbResponse(FrontendHbResponse hbResponse, boolean isReplay)
             } else {
                 if (isAlive) {
                     isAlive = false;
-                    isChanged = true;
                 }
                 heartbeatErrMsg = hbResponse.getMsg() == null ? "Unknown error" : hbResponse.getMsg();
             }
+            // When the master receives an error heartbeat info which status not ok, 
+            // this heartbeat info also need to be synced to follower.
+            // Since the failed heartbeat info also modifies fe's memory, (this.heartbeatRetryTimes++;)
+            // if this heartbeat is not synchronized to the follower, 
+            // that will cause the Follower and master’s memory to be inconsistent
+            isChanged = true;
         }
         return isChanged;
     }

fe/fe-core/src/test/java/com/starrocks/system/ComputeNodeTest.java

@@ -0,0 +1,21 @@
+// This file is licensed under the Elastic License 2.0. Copyright 2021-present, StarRocks Limited.
+package com.starrocks.system;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.starrocks.system.HeartbeatResponse.HbStatus;
+
+public class ComputeNodeTest {
+
+    @Test
+    public void testHbStatusBadNeedSync() {
+
+        BackendHbResponse hbResponse = new BackendHbResponse();
+        hbResponse.status = HbStatus.BAD;
+
+        ComputeNode node = new ComputeNode();
+        boolean needSync = node.handleHbResponse(hbResponse);
+        Assert.assertTrue(needSync);
+    }
+}

fe/fe-core/src/test/java/com/starrocks/system/FrontendTest.java

@@ -16,4 +16,13 @@ public void testFeUpdate() {
         Assert.assertEquals("modifiedHost", fe.getHost());
         Assert.assertTrue(fe.getEditLogPort() == 2110);
     }
+
+    @Test
+    public void testHbStatusBadNeedSync() {
+        FrontendHbResponse hbResponse = new FrontendHbResponse("BAD", "");
+
+        Frontend fe = new Frontend(FrontendNodeType.FOLLOWER, "name", "testHost", 1110);
+        boolean needSync = fe.handleHbResponse(hbResponse, true);
+        Assert.assertTrue(needSync);
+    }
 }