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

…ynced to follower (#8265)

When master the received heartbeat status is not ok, the heartbeat info also need to be synced to follower.
Otherwise, the failed heartbeat information will not be synchronized to the follower.
Since the failed heartbeat info also modifies fe's memory, this.heartbeatRetryTimes++;
if it is not synchronized to the follower, this will cause the master and follower's metadata to be inconsistent.

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