[ISSUE #247]revote when leader's term is not the biggest

Co-authored-by: dengzhiwen1 <dengzhiwen1@xiaomi.com>

Author: cserwen

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

@@ -304,16 +304,24 @@ private void sendHeartbeats(long term, String leaderId) throws Exception {
                 }
             });
         }
-        beatLatch.await(heartBeatTimeIntervalMs, TimeUnit.MILLISECONDS);
+        long voteResultWaitTime = 10;
+        beatLatch.await(heartBeatTimeIntervalMs - voteResultWaitTime, TimeUnit.MILLISECONDS);
+        Thread.sleep(voteResultWaitTime);
+
+        //abnormal case, deal with it immediately
+        if (maxTerm.get() > term) {
+            LOGGER.warn("[{}] currentTerm{} is not the biggest={}, deal with it", memberState.getSelfId(), term, maxTerm.get());
+            changeRoleToCandidate(maxTerm.get());
+            return;
+        }
+
         if (memberState.isQuorum(succNum.get())) {
             lastSuccHeartBeatTime = System.currentTimeMillis();
         } 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));
             if (memberState.isQuorum(succNum.get() + notReadyNum.get())) {
                 lastSendHeartBeatTime = -1;
-            } else if (maxTerm.get() > term) {
-                changeRoleToCandidate(maxTerm.get());
             } else if (inconsistLeader.get()) {
                 changeRoleToCandidate(term);
             } else if (DLedgerUtils.elapsed(lastSuccHeartBeatTime) > (long) maxHeartBeatLeak * heartBeatTimeIntervalMs) {

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

@@ -367,6 +367,7 @@ public CompletableFuture<PushEntryResponse> handlePush(PushEntryRequest request)
     @Override
     public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(
         LeadershipTransferRequest request) throws Exception {
+        LOGGER.info("handleLeadershipTransfer: {}", request);
         try {
             PreConditions.check(memberState.getSelfId().equals(request.getRemoteId()), DLedgerResponseCode.UNKNOWN_MEMBER, "%s != %s", request.getRemoteId(), memberState.getSelfId());
             PreConditions.check(memberState.getGroup().equals(request.getGroup()), DLedgerResponseCode.UNKNOWN_GROUP, "%s != %s", request.getGroup(), memberState.getGroup());