You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by eo...@apache.org on 2022/09/29 13:39:30 UTC
[bookkeeper] branch master updated: Issue 3070: Fix bug where checkAllLedgers gets stuck when read throttling is enabled (#3214)
This is an automated email from the ASF dual-hosted git repository.
eolivelli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 21560fb9af Issue 3070: Fix bug where checkAllLedgers gets stuck when read throttling is enabled (#3214)
21560fb9af is described below
commit 21560fb9afd9da61231a709c0b86a1a7b2d0f8b0
Author: Masahiro Sakamoto <ma...@yahoo-corp.jp>
AuthorDate: Thu Sep 29 22:39:24 2022 +0900
Issue 3070: Fix bug where checkAllLedgers gets stuck when read throttling is enabled (#3214)
---
.../main/java/org/apache/bookkeeper/client/LedgerChecker.java | 9 +++++++--
.../java/org/apache/bookkeeper/client/TestLedgerChecker.java | 2 +-
2 files changed, 8 insertions(+), 3 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerChecker.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerChecker.java
index d943e05899..b9df5f4a22 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerChecker.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerChecker.java
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -426,6 +427,11 @@ public class LedgerChecker {
if (curEntryId == lastEntry) {
final long entryToRead = curEntryId;
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ future.whenCompleteAsync((re, ex) -> {
+ checkFragments(fragments, cb, percentageOfLedgerFragmentToBeVerified);
+ });
+
final EntryExistsCallback eecb = new EntryExistsCallback(lh.getLedgerMetadata().getWriteQuorumSize(),
new GenericCallback<Boolean>() {
@Override
@@ -433,8 +439,7 @@ public class LedgerChecker {
if (result) {
fragments.add(lastLedgerFragment);
}
- checkFragments(fragments, cb,
- percentageOfLedgerFragmentToBeVerified);
+ future.complete(null);
}
});
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerChecker.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerChecker.java
index 4fa3797856..4f3ee9fc9c 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerChecker.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerChecker.java
@@ -517,7 +517,7 @@ public class TestLedgerChecker extends BookKeeperClusterTestCase {
private Set<LedgerFragment> getUnderReplicatedFragments(LedgerHandle lh)
throws InterruptedException {
- LedgerChecker checker = new LedgerChecker(bkc);
+ LedgerChecker checker = new LedgerChecker(bkc, 1);
CheckerCallback cb = new CheckerCallback();
checker.checkLedger(lh, cb);
Set<LedgerFragment> result = cb.waitAndGetResult();