[ISSUE #85] fix preferred leader bug

Signed-off-by: zhangyang21 <zhangyang21@xiaomi.com>
openmessaging · Jan 26, 2022 · efaddd3 · efaddd3
1 parent e008ef5
commit efaddd3
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 1 deletion.
diff --git a/src/main/java/io/openmessaging/storage/dledger/DLedgerConfig.java b/src/main/java/io/openmessaging/storage/dledger/DLedgerConfig.java
@@ -88,6 +88,7 @@ public class DLedgerConfig {
     private boolean isEnableBatchPush = false;
     private int maxBatchPushSize = 4 * 1024;
 
+    private long leadershipTransferWaitTimeout = 1000;
 
     public String getDefaultPath() {
         return storeBaseDir + File.separator + "dledger-" + selfId;
@@ -398,4 +399,12 @@ public int getMaxBatchPushSize() {
     public void setMaxBatchPushSize(int maxBatchPushSize) {
         this.maxBatchPushSize = maxBatchPushSize;
     }
+
+    public long getLeadershipTransferWaitTimeout() {
+        return leadershipTransferWaitTimeout;
+    }
+
+    public void setLeadershipTransferWaitTimeout(long leadershipTransferWaitTimeout) {
+        this.leadershipTransferWaitTimeout = leadershipTransferWaitTimeout;
+    }
 }
diff --git a/src/main/java/io/openmessaging/storage/dledger/DLedgerServer.java b/src/main/java/io/openmessaging/storage/dledger/DLedgerServer.java
@@ -296,7 +296,16 @@ public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(Le
             } else if (memberState.getSelfId().equals(request.getTransfereeId())) {
                 // It's the transferee received the take leadership command.
                 PreConditions.check(request.getTransferId().equals(memberState.getLeaderId()), DLedgerResponseCode.INCONSISTENT_LEADER, "transfer=%s is not leader", request.getTransferId());
-
+                long startTime = System.currentTimeMillis();
+                long fallBehind;
+                while ((fallBehind = request.getTakeLeadershipLedgerIndex() - memberState.getLedgerEndIndex()) > 0) {
+                    if (System.currentTimeMillis() - startTime > dLedgerConfig.getLeadershipTransferWaitTimeout()) {
+                        logger.warn("transferee fall behind, wait timeout, diff={}.", fallBehind);
+                        break;
+                    }
+                    logger.warn("transferee fall behind, wait 1ms, diff={}.", fallBehind);
+                    Thread.sleep(1);
+                }
                 return dLedgerLeaderElector.handleTakeLeadership(request);
             } else {
                 return CompletableFuture.completedFuture(new LeadershipTransferResponse().term(memberState.currTerm()).code(DLedgerResponseCode.UNEXPECTED_ARGUMENT.getCode()));