fix ha sync transfer timeout

85a6cc1c · wqliang · 7c469fe9 · 85a6cc1c
显示空白变更内容
内联并排

Showing with 6 addition and 1 deletion

store/src/main/java/org/apache/rocketmq/store/ha/HAService.java ...src/main/java/org/apache/rocketmq/store/ha/HAService.java +6 -1

未找到文件。
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
@@ -280,9 +280,14 @@ public class HAService {
                if (!this.requestsRead.isEmpty()) {
                    for (CommitLog.GroupCommitRequest req : this.requestsRead) {
                        boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
-                        for (int i = 0; !transferOK && i < 5; i++) {
+                        long waitUntillWhen = HAService.this.defaultMessageStore.getSystemClock().now()
+                            + HAService.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout();
+                        while (HAService.this.defaultMessageStore.getSystemClock().now() < waitUntillWhen) {
                            this.notifyTransferObject.waitForRunning(1000);
                            transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
+                            if (transferOK) {
+                                break;
+                            }
                        }

                        if (!transferOK) {