(bookkeeper) branch branch-4.17 updated: [fix] Ledger Replication worker should throttle on verrification reads (#4678)

[chenhang]

This is an automated email from the ASF dual-hosted git repository.

hangc0276 pushed a commit to branch branch-4.17
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git


The following commit(s) were added to refs/heads/branch-4.17 by this push:
     new c9361a0e3c [fix] Ledger Replication worker should throttle on 
verrification reads (#4678)
c9361a0e3c is described below

commit c9361a0e3ce2ae67d0327b4851d949aa37d78fe1
Author: vraulji567 <95091480+vraulji...@users.noreply.github.com>
AuthorDate: Tue May 26 20:43:17 2026 -0400

    [fix] Ledger Replication worker should throttle on verrification reads 
(#4678)
---
 .../bookkeeper/replication/ReplicationWorker.java  |  2 +-
 .../bookkeeper/client/TestLedgerChecker.java       | 51 ++++++++++++++++++++++
 2 files changed, 52 insertions(+), 1 deletion(-)

diff --git 
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java
 
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java
index 05807455dc..cebd8a1b55 100644
--- 
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java
+++ 
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java
@@ -192,7 +192,7 @@ public class ReplicationWorker implements Runnable {
         this.underreplicationManager = 
bkc.getLedgerManagerFactory().newLedgerUnderreplicationManager();
         this.ledgerManager = bkc.getLedgerManagerFactory().newLedgerManager();
         this.admin = new BookKeeperAdmin(bkc, statsLogger, new 
ClientConfiguration(conf));
-        this.ledgerChecker = new LedgerChecker(bkc);
+        this.ledgerChecker = new LedgerChecker(bkc, 
conf.getInFlightReadEntryNumInLedgerChecker());
         this.workerThread = new BookieThread(this, "ReplicationWorker");
         this.openLedgerRereplicationGracePeriod = conf
                 .getOpenLedgerRereplicationGracePeriod();
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 222cd0aac4..545c9b8ac0 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
@@ -24,6 +24,9 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -34,6 +37,8 @@ import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.proto.BookieClient;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
 import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
 import org.junit.Test;
@@ -530,4 +535,50 @@ public class TestLedgerChecker extends 
BookKeeperClusterTestCase {
         killBookie(ensemble);
     }
 
+    @Test
+    public void testLedgerCheckerDoesNotExceedPermits() throws Exception {
+        int permits = 100;
+
+        // Atomic counter to track concurrent reads
+        AtomicInteger concurrentReads = new AtomicInteger(0);
+        AtomicInteger maxConcurrentReads = new AtomicInteger(0);
+
+        // Mock Bookie Client
+        BookieClient mockBookieClient = mock(BookieClient.class);
+        BookieWatcher mockBookieWatcher = mock(BookieWatcher.class);
+        when(mockBookieWatcher.isBookieUnavailable(any())).thenReturn(false);
+
+        doAnswer(invocation -> {
+            int now = concurrentReads.incrementAndGet();
+            maxConcurrentReads.updateAndGet(prev -> Math.max(prev, now));
+
+            // Simulate async completion
+            new Thread(() -> {
+                try{Thread.sleep(10);} catch (InterruptedException ie){}
+                int d = concurrentReads.decrementAndGet();
+                BookkeeperInternalCallbacks.ReadEntryCallback cb = 
invocation.getArgument(3);
+                cb.readEntryComplete(0, invocation.getArgument(1),
+                        invocation.getArgument(2), null, null);
+            }).start();
+            return null;
+        }).when(mockBookieClient).readEntry(any(BookieId.class), anyLong(), 
anyLong(),
+                any(), any(), anyInt());
+
+        // Create dummy LedgerHandle with enough fragments
+        LedgerHandle lh = bkc.createLedger(BookKeeper.DigestType.CRC32, 
TEST_LEDGER_PASSWORD);
+        startNewBookie();
+        for(int i = 0; i < 1000; i++) {
+            lh.addEntry(TEST_LEDGER_ENTRY_DATA);
+        }
+
+        LedgerChecker checker = new LedgerChecker(mockBookieClient, 
mockBookieWatcher, permits);
+        checker.checkLedger(lh, (rc, result) -> {}, 100);
+
+        // Wait for all async reads to finish
+        Thread.sleep(2000);
+
+        System.out.println("Max Concurrent reads observed = " + 
maxConcurrentReads.get());
+        assertTrue("Exceeded permit limit!", maxConcurrentReads.get() <= 
permits);
+    }
+
 }