[ISSUE #144] node abnormal status detection and recovery

Signed-off-by: zhangyang21 <zhangyang21@xiaomi.com>

Author: Git-Yang

src/main/java/io/openmessaging/storage/dledger/DLedgerLeaderElector.java

@@ -29,6 +29,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
@@ -111,9 +112,7 @@ public CompletableFuture<HeartBeatResponse> handleHeartBeat(HeartBeatRequest req
             return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.UNEXPECTED_MEMBER.getCode()));
         }
 
-        if (request.getTerm() < memberState.currTerm()) {
-            return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
-        } else if (request.getTerm() == memberState.currTerm()) {
+        if (request.getTerm() == memberState.currTerm()) {
             if (request.getLeaderId().equals(memberState.getLeaderId())) {
                 lastLeaderHeartBeatTime = System.currentTimeMillis();
                 return CompletableFuture.completedFuture(new HeartBeatResponse());
@@ -124,8 +123,15 @@ public CompletableFuture<HeartBeatResponse> handleHeartBeat(HeartBeatRequest req
         //hold the lock to get the latest term and leaderId
         synchronized (memberState) {
             if (request.getTerm() < memberState.currTerm()) {
+                if (memberState.isCandidate() && request.isNeedCheckMemberState()) {
+                    logger.warn("[CHECK_MEMBER_STATE] [HandleHeartBeat] remoteId={} need check member state", request.getLeaderId());
+                    memberState.recoveryToFollower(request.getTerm(), request.getLeaderId());
+                    return CompletableFuture.completedFuture(new HeartBeatResponse());
+                }
                 return CompletableFuture.completedFuture(new HeartBeatResponse().term(memberState.currTerm()).code(DLedgerResponseCode.EXPIRED_TERM.getCode()));
-            } else if (request.getTerm() == memberState.currTerm()) {
+            }
+
+            if (request.getTerm() == memberState.currTerm()) {
                 if (memberState.getLeaderId() == null) {
                     changeRoleToFollower(request.getTerm(), request.getLeaderId());
                     return CompletableFuture.completedFuture(new HeartBeatResponse());
@@ -283,10 +289,12 @@ private void sendHeartbeats(long term, String leaderId) throws Exception {
                             break;
                     }
 
-                    if (x.getCode() == DLedgerResponseCode.NETWORK_ERROR.getCode())
+                    if (x.getCode() == DLedgerResponseCode.NETWORK_ERROR.getCode()) {
                         memberState.getPeersLiveTable().put(id, Boolean.FALSE);
-                    else
+                    } else {
                         memberState.getPeersLiveTable().put(id, Boolean.TRUE);
+                        memberState.getPeersTermTable().put(id, x.getTerm());
+                    }
 
                     if (memberState.isQuorum(succNum.get())
                         || memberState.isQuorum(succNum.get() + notReadyNum.get())) {
@@ -305,6 +313,7 @@ private void sendHeartbeats(long term, String leaderId) throws Exception {
         beatLatch.await(heartBeatTimeIntervalMs, TimeUnit.MILLISECONDS);
         if (memberState.isQuorum(succNum.get())) {
             lastSuccHeartBeatTime = System.currentTimeMillis();
+            checkPeersTermTable();
         } else {
             logger.info("[{}] Parse heartbeat responses in cost={} term={} allNum={} succNum={} notReadyNum={} inconsistLeader={} maxTerm={} peerSize={} lastSuccHeartBeatTime={}",
                 memberState.getSelfId(), DLedgerUtils.elapsed(startHeartbeatTimeMs), term, allNum.get(), succNum.get(), notReadyNum.get(), inconsistLeader.get(), maxTerm.get(), memberState.peerSize(), new Timestamp(lastSuccHeartBeatTime));
@@ -320,6 +329,28 @@ private void sendHeartbeats(long term, String leaderId) throws Exception {
         }
     }
 
+    private void checkPeersTermTable() throws Exception {
+        if (memberState.getSelfId().equals(memberState.getLeaderId())) {
+            long leaderTerm = memberState.getPeersTermTable().getOrDefault(memberState.getLeaderId(), -1L);
+            for (Map.Entry<String, Long> entryTerm : memberState.getPeersTermTable().entrySet()) {
+                if (entryTerm.getKey().equals(memberState.getSelfId())) {
+                    continue;
+                }
+
+                if (entryTerm.getValue() > leaderTerm) {
+                    HeartBeatRequest heartBeatRequest = new HeartBeatRequest();
+                    heartBeatRequest.setGroup(memberState.getGroup());
+                    heartBeatRequest.setLocalId(memberState.getSelfId());
+                    heartBeatRequest.setRemoteId(entryTerm.getKey());
+                    heartBeatRequest.setLeaderId(memberState.getLeaderId());
+                    heartBeatRequest.setNeedCheckMemberState(true);
+                    heartBeatRequest.setTerm(leaderTerm);
+                    dLedgerRpcService.heartBeat(heartBeatRequest);
+                }
+            }
+        }
+    }
+
     private void maintainAsLeader() throws Exception {
         if (DLedgerUtils.elapsed(lastSendHeartBeatTime) > heartBeatTimeIntervalMs) {
             long term;

src/main/java/io/openmessaging/storage/dledger/MemberState.java

@@ -52,6 +52,7 @@ public class MemberState {
     private long knownMaxTermInGroup = -1;
     private Map<String, String> peerMap = new HashMap<>();
     private Map<String, Boolean> peersLiveTable = new ConcurrentHashMap<>();
+    private Map<String, Long> peersTermTable = new HashMap<>();
 
     private volatile String transferee;
     private volatile long termToTakeLeadership = -1;
@@ -132,6 +133,7 @@ public synchronized void changeToLeader(long term) {
         this.role = LEADER;
         this.leaderId = selfId;
         peersLiveTable.clear();
+        peersTermTable.clear();
     }
 
     public synchronized void changeToFollower(long term, String leaderId) {
@@ -153,6 +155,13 @@ public synchronized void changeToCandidate(long term) {
         transferee = null;
     }
 
+    public synchronized void recoveryToFollower(long term, String leaderId) {
+        this.role = FOLLOWER;
+        this.leaderId = leaderId;
+        this.currTerm = term;
+        transferee = null;
+    }
+
     public String getTransferee() {
         return transferee;
     }
@@ -226,6 +235,10 @@ public Map<String, Boolean> getPeersLiveTable() {
         return peersLiveTable;
     }
 
+    public Map<String, Long> getPeersTermTable() {
+        return peersTermTable;
+    }
+
     //just for test
     public void setCurrTermForTest(long term) {
         PreConditions.check(term >= currTerm, DLedgerResponseCode.ILLEGAL_MEMBER_STATE);

src/main/java/io/openmessaging/storage/dledger/protocol/HeartBeatRequest.java

@@ -18,4 +18,13 @@
 
 public class HeartBeatRequest extends RequestOrResponse {
 
+    private boolean needCheckMemberState = false;
+
+    public boolean isNeedCheckMemberState() {
+        return needCheckMemberState;
+    }
+
+    public void setNeedCheckMemberState(boolean needCheckMemberState) {
+        this.needCheckMemberState = needCheckMemberState;
+    }
 }