From 5a5e79ffbf1dc5844d835f8b9c16ff175e616c2a Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Wed, 24 Jun 2026 16:43:42 +0800 Subject: [PATCH 1/3] Fix managed ledger read timeout tracking --- .../mledger/impl/ManagedLedgerImpl.java | 55 +++++++++------- .../mledger/impl/ManagedLedgerTest.java | 62 +++++++++++++++++++ 2 files changed, 95 insertions(+), 22 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 22d1655b718ec..6e75fa150f718 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -348,11 +348,8 @@ public boolean isFenced() { .newUpdater(ManagedLedgerImpl.class, "addOpCount"); private volatile long addOpCount = 0; - // last read-operation's callback to check read-timeout on it. - private volatile ReadEntryCallbackWrapper lastReadCallback = null; - private static final AtomicReferenceFieldUpdater - LAST_READ_CALLBACK_UPDATER = AtomicReferenceFieldUpdater - .newUpdater(ManagedLedgerImpl.class, ReadEntryCallbackWrapper.class, "lastReadCallback"); + // Pending read callbacks tracked by read operation id for read-timeout checks. + private final ConcurrentHashMap pendingReadCallbacks = new ConcurrentHashMap<>(); /** * Queue of pending entries to be added to the managed ledger. Typically, entries are queued when a new ledger is. @@ -2439,8 +2436,8 @@ protected void asyncReadEntry(ReadHandle ledger, Position position, ReadEntryCal long readOpCount = READ_OP_COUNT_UPDATER.incrementAndGet(this); long createdTime = System.nanoTime(); ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, position.getLedgerId(), - position.getEntryId(), callback, readOpCount, createdTime, ctx); - lastReadCallback = readCallback; + position.getEntryId(), callback, readOpCount, createdTime, ctx, pendingReadCallbacks); + pendingReadCallbacks.put(readOpCount, readCallback); entryCache.asyncReadEntry(ledger, position, readCallback, readOpCount); } else { entryCache.asyncReadEntry(ledger, position, callback, ctx); @@ -2455,8 +2452,8 @@ protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry long readOpCount = READ_OP_COUNT_UPDATER.incrementAndGet(this); long createdTime = System.nanoTime(); ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, ledger.getId(), firstEntry, - opReadEntry, readOpCount, createdTime, ctx); - lastReadCallback = readCallback; + opReadEntry, readOpCount, createdTime, ctx, pendingReadCallbacks); + pendingReadCallbacks.put(readOpCount, readCallback); entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, readCallback, readOpCount); } else { entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, opReadEntry, ctx); @@ -2475,6 +2472,7 @@ static final class ReadEntryCallbackWrapper implements ReadEntryCallback, ReadEn AtomicLongFieldUpdater.newUpdater(ReadEntryCallbackWrapper.class, "readOpCount"); volatile long createdTime = -1; volatile Object cntx; + volatile ConcurrentHashMap pendingReadCallbacks; final Handle recyclerHandle; @@ -2483,7 +2481,8 @@ private ReadEntryCallbackWrapper(Handle recyclerHandle } static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntryCallback callback, - long readOpCount, long createdTime, Object ctx) { + long readOpCount, long createdTime, Object ctx, + ConcurrentHashMap pendingReadCallbacks) { ReadEntryCallbackWrapper readCallback = RECYCLER.get(); readCallback.name = name; readCallback.ledgerId = ledgerId; @@ -2492,11 +2491,13 @@ static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, readCallback.cntx = ctx; readCallback.readOpCount = readOpCount; readCallback.createdTime = createdTime; + readCallback.pendingReadCallbacks = pendingReadCallbacks; return readCallback; } static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntriesCallback callback, - long readOpCount, long createdTime, Object ctx) { + long readOpCount, long createdTime, Object ctx, + ConcurrentHashMap pendingReadCallbacks) { ReadEntryCallbackWrapper readCallback = RECYCLER.get(); readCallback.name = name; readCallback.ledgerId = ledgerId; @@ -2505,6 +2506,7 @@ static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, readCallback.cntx = ctx; readCallback.readOpCount = readOpCount; readCallback.createdTime = createdTime; + readCallback.pendingReadCallbacks = pendingReadCallbacks; return readCallback; } @@ -2575,12 +2577,18 @@ public void readFailed(ManagedLedgerException exception, Object ctx) { private boolean recycle(long readOpCount) { if (readOpCount != -1 && READ_OP_COUNT_UPDATER.compareAndSet(ReadEntryCallbackWrapper.this, readOpCount, -1)) { + ConcurrentHashMap pendingCallbacks = pendingReadCallbacks; + if (pendingCallbacks != null) { + pendingCallbacks.remove(readOpCount, this); + } createdTime = -1; readEntryCallback = null; readEntriesCallback = null; ledgerId = -1; entryId = -1; name = null; + cntx = null; + pendingReadCallbacks = null; recyclerHandle.recycle(this); return true; } @@ -4771,17 +4779,20 @@ private void checkReadTimeout() { if (timeoutSec < 1) { return; } - ReadEntryCallbackWrapper callback = this.lastReadCallback; - long readOpCount = callback != null ? callback.readOpCount : 0; - boolean timeout = callback != null && (TimeUnit.NANOSECONDS - .toSeconds(System.nanoTime() - callback.createdTime) >= timeoutSec); - if (readOpCount > 0 && timeout) { - log.warn().attr("ledgerId", this.lastReadCallback.ledgerId) - .attr("entryId", this.lastReadCallback.entryId) - .attr("timeoutSec", timeoutSec).log("Read entry timeout"); - callback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException), readOpCount); - LAST_READ_CALLBACK_UPDATER.compareAndSet(this, callback, null); - } + long now = System.nanoTime(); + pendingReadCallbacks.forEach((readOpCount, callback) -> { + if (readOpCount <= 0 || callback == null || callback.createdTime < 0) { + return; + } + boolean timeout = TimeUnit.NANOSECONDS.toSeconds(now - callback.createdTime) >= timeoutSec; + if (timeout && pendingReadCallbacks.remove(readOpCount, callback)) { + log.warn().attr("ledgerId", callback.ledgerId) + .attr("entryId", callback.entryId) + .attr("timeoutSec", timeoutSec) + .log("Read entry timeout"); + callback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException), readOpCount); + } + }); } @Override diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index 5dbdcaa71e8b5..b6979002bc655 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -3427,6 +3427,68 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { ledger.close(); } + @Test + public void testManagedLedgerWithConcurrentReadEntryTimeOut() throws Exception { + ManagedLedgerConfig config = initManagedLedgerConfig(new ManagedLedgerConfig()).setReadEntryTimeoutSeconds(1); + ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("concurrent_timeout_ledger_test", config); + + Position position1 = ledger.addEntry("entry-1".getBytes()); + Position position2 = ledger.addEntry("entry-2".getBytes()); + + // ensure that the reads aren't cached + factory.getEntryCacheManager().clear(); + + bkc.setReadHandleInterceptor(new PulsarMockReadHandleInterceptor() { + @Override + public CompletableFuture interceptReadAsync(long ledgerId, long firstEntry, long lastEntry, + LedgerEntries entries) { + return CompletableFuture.supplyAsync(() -> entries, + CompletableFuture.delayedExecutor(3, TimeUnit.SECONDS)); + } + }); + + AtomicReference responseException1 = new AtomicReference<>(); + AtomicReference responseException2 = new AtomicReference<>(); + String ctxStr = "timeoutCtx"; + + ledger.asyncReadEntry(position1, new ReadEntryCallback() { + @Override + public void readEntryComplete(Entry entry, Object ctx) { + entry.release(); + } + + @Override + public void readEntryFailed(ManagedLedgerException exception, Object ctx) { + assertEquals(ctxStr, (String) ctx); + responseException1.set(exception); + } + }, ctxStr); + + ledger.asyncReadEntry(position2, new ReadEntryCallback() { + @Override + public void readEntryComplete(Entry entry, Object ctx) { + entry.release(); + } + + @Override + public void readEntryFailed(ManagedLedgerException exception, Object ctx) { + assertEquals(ctxStr, (String) ctx); + responseException2.set(exception); + } + }, ctxStr); + + Awaitility.await().untilAsserted(() -> { + assertNotNull(responseException1.get()); + assertTrue(responseException1.get().getMessage() + .startsWith(BKException.getMessage(BKException.Code.TimeoutException))); + assertNotNull(responseException2.get()); + assertTrue(responseException2.get().getMessage() + .startsWith(BKException.getMessage(BKException.Code.TimeoutException))); + }); + + ledger.close(); + } + @Test public void testAddEntryResponseTimeout() throws Exception { // Create ML with feature Add Entry Timeout Check. From 9ba177d3fe94963adb9441921f6251fc15fb86f9 Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Thu, 25 Jun 2026 10:12:11 +0800 Subject: [PATCH 2/3] Use priority queue for read timeouts --- .../mledger/impl/ManagedLedgerImpl.java | 107 ++++++++++++------ 1 file changed, 75 insertions(+), 32 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 6e75fa150f718..01d5d20065337 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -56,6 +56,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -348,8 +349,8 @@ public boolean isFenced() { .newUpdater(ManagedLedgerImpl.class, "addOpCount"); private volatile long addOpCount = 0; - // Pending read callbacks tracked by read operation id for read-timeout checks. - private final ConcurrentHashMap pendingReadCallbacks = new ConcurrentHashMap<>(); + // Pending reads ordered by timeout so checks can stop at the first operation that has not expired. + private final PriorityBlockingQueue pendingReadCallbacks = new PriorityBlockingQueue<>(); /** * Queue of pending entries to be added to the managed ledger. Typically, entries are queued when a new ledger is. @@ -2434,10 +2435,13 @@ protected void asyncReadEntry(ReadHandle ledger, Position position, ReadEntryCal if (config.getReadEntryTimeoutSeconds() > 0) { // set readOpCount to uniquely validate if ReadEntryCallbackWrapper is already recycled long readOpCount = READ_OP_COUNT_UPDATER.incrementAndGet(this); - long createdTime = System.nanoTime(); ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, position.getLedgerId(), - position.getEntryId(), callback, readOpCount, createdTime, ctx, pendingReadCallbacks); - pendingReadCallbacks.put(readOpCount, readCallback); + position.getEntryId(), callback, readOpCount, ctx); + PendingReadCallback pendingReadCallback = new PendingReadCallback(readOpCount, + timeoutAtNanos(config.getReadEntryTimeoutSeconds()), position.getLedgerId(), position.getEntryId(), + readCallback); + readCallback.pendingReadCallback = pendingReadCallback; + pendingReadCallbacks.add(pendingReadCallback); entryCache.asyncReadEntry(ledger, position, readCallback, readOpCount); } else { entryCache.asyncReadEntry(ledger, position, callback, ctx); @@ -2450,16 +2454,57 @@ protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry if (config.getReadEntryTimeoutSeconds() > 0) { // set readOpCount to uniquely validate if ReadEntryCallbackWrapper is already recycled long readOpCount = READ_OP_COUNT_UPDATER.incrementAndGet(this); - long createdTime = System.nanoTime(); ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, ledger.getId(), firstEntry, - opReadEntry, readOpCount, createdTime, ctx, pendingReadCallbacks); - pendingReadCallbacks.put(readOpCount, readCallback); + opReadEntry, readOpCount, ctx); + PendingReadCallback pendingReadCallback = new PendingReadCallback(readOpCount, + timeoutAtNanos(config.getReadEntryTimeoutSeconds()), ledger.getId(), firstEntry, readCallback); + readCallback.pendingReadCallback = pendingReadCallback; + pendingReadCallbacks.add(pendingReadCallback); entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, readCallback, readOpCount); } else { entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, opReadEntry, ctx); } } + private static long timeoutAtNanos(long timeoutSec) { + return System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSec); + } + + static final class PendingReadCallback implements Comparable { + private static final AtomicReferenceFieldUpdater + CALLBACK_UPDATER = AtomicReferenceFieldUpdater + .newUpdater(PendingReadCallback.class, ReadEntryCallbackWrapper.class, "callback"); + + final long readOpCount; + final long timeoutAtNanos; + final long ledgerId; + final long entryId; + volatile ReadEntryCallbackWrapper callback; + + PendingReadCallback(long readOpCount, long timeoutAtNanos, long ledgerId, long entryId, + ReadEntryCallbackWrapper callback) { + this.readOpCount = readOpCount; + this.timeoutAtNanos = timeoutAtNanos; + this.ledgerId = ledgerId; + this.entryId = entryId; + this.callback = callback; + } + + @Override + public int compareTo(PendingReadCallback other) { + int result = Long.compare(timeoutAtNanos, other.timeoutAtNanos); + return result != 0 ? result : Long.compare(readOpCount, other.readOpCount); + } + + ReadEntryCallbackWrapper clearCallback() { + return CALLBACK_UPDATER.getAndSet(this, null); + } + + void clearCallback(ReadEntryCallbackWrapper callback) { + CALLBACK_UPDATER.compareAndSet(this, callback, null); + } + } + static final class ReadEntryCallbackWrapper implements ReadEntryCallback, ReadEntriesCallback { volatile ReadEntryCallback readEntryCallback; @@ -2470,9 +2515,8 @@ static final class ReadEntryCallbackWrapper implements ReadEntryCallback, ReadEn volatile long readOpCount = -1; private static final AtomicLongFieldUpdater READ_OP_COUNT_UPDATER = AtomicLongFieldUpdater.newUpdater(ReadEntryCallbackWrapper.class, "readOpCount"); - volatile long createdTime = -1; volatile Object cntx; - volatile ConcurrentHashMap pendingReadCallbacks; + volatile PendingReadCallback pendingReadCallback; final Handle recyclerHandle; @@ -2481,8 +2525,7 @@ private ReadEntryCallbackWrapper(Handle recyclerHandle } static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntryCallback callback, - long readOpCount, long createdTime, Object ctx, - ConcurrentHashMap pendingReadCallbacks) { + long readOpCount, Object ctx) { ReadEntryCallbackWrapper readCallback = RECYCLER.get(); readCallback.name = name; readCallback.ledgerId = ledgerId; @@ -2490,14 +2533,11 @@ static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, readCallback.readEntryCallback = callback; readCallback.cntx = ctx; readCallback.readOpCount = readOpCount; - readCallback.createdTime = createdTime; - readCallback.pendingReadCallbacks = pendingReadCallbacks; return readCallback; } static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntriesCallback callback, - long readOpCount, long createdTime, Object ctx, - ConcurrentHashMap pendingReadCallbacks) { + long readOpCount, Object ctx) { ReadEntryCallbackWrapper readCallback = RECYCLER.get(); readCallback.name = name; readCallback.ledgerId = ledgerId; @@ -2505,8 +2545,6 @@ static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, readCallback.readEntriesCallback = callback; readCallback.cntx = ctx; readCallback.readOpCount = readOpCount; - readCallback.createdTime = createdTime; - readCallback.pendingReadCallbacks = pendingReadCallbacks; return readCallback; } @@ -2577,18 +2615,17 @@ public void readFailed(ManagedLedgerException exception, Object ctx) { private boolean recycle(long readOpCount) { if (readOpCount != -1 && READ_OP_COUNT_UPDATER.compareAndSet(ReadEntryCallbackWrapper.this, readOpCount, -1)) { - ConcurrentHashMap pendingCallbacks = pendingReadCallbacks; - if (pendingCallbacks != null) { - pendingCallbacks.remove(readOpCount, this); + PendingReadCallback pendingCallback = pendingReadCallback; + if (pendingCallback != null) { + pendingCallback.clearCallback(this); } - createdTime = -1; readEntryCallback = null; readEntriesCallback = null; ledgerId = -1; entryId = -1; name = null; cntx = null; - pendingReadCallbacks = null; + pendingReadCallback = null; recyclerHandle.recycle(this); return true; } @@ -4780,19 +4817,25 @@ private void checkReadTimeout() { return; } long now = System.nanoTime(); - pendingReadCallbacks.forEach((readOpCount, callback) -> { - if (readOpCount <= 0 || callback == null || callback.createdTime < 0) { - return; + while (true) { + PendingReadCallback pendingCallback = pendingReadCallbacks.peek(); + if (pendingCallback == null || pendingCallback.timeoutAtNanos > now) { + break; + } + pendingCallback = pendingReadCallbacks.poll(); + if (pendingCallback == null) { + break; } - boolean timeout = TimeUnit.NANOSECONDS.toSeconds(now - callback.createdTime) >= timeoutSec; - if (timeout && pendingReadCallbacks.remove(readOpCount, callback)) { - log.warn().attr("ledgerId", callback.ledgerId) - .attr("entryId", callback.entryId) + ReadEntryCallbackWrapper callback = pendingCallback.clearCallback(); + if (callback != null) { + log.warn().attr("ledgerId", pendingCallback.ledgerId) + .attr("entryId", pendingCallback.entryId) .attr("timeoutSec", timeoutSec) .log("Read entry timeout"); - callback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException), readOpCount); + callback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException), + pendingCallback.readOpCount); } - }); + } } @Override From d9e6e494633c01f6cce37fbf033eaafa2760ee8f Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Mon, 31 Aug 2026 18:13:13 +0800 Subject: [PATCH 3/3] [fix][ml] Share read timeout tracking --- .../impl/ManagedLedgerFactoryImpl.java | 7 + .../mledger/impl/ManagedLedgerImpl.java | 139 +++++------------- .../mledger/impl/ReadEntryTimeoutTracker.java | 132 +++++++++++++++++ .../mledger/impl/ManagedLedgerTest.java | 27 ++++ 4 files changed, 200 insertions(+), 105 deletions(-) create mode 100644 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadEntryTimeoutTracker.java diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java index 269cbafc20799..4c2095dc5339a 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java @@ -133,6 +133,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory { private long lastStatTimestamp = System.nanoTime(); private final ScheduledFuture statsTask; private final ScheduledFuture flushCursorsTask; + private final ReadEntryTimeoutTracker readEntryTimeoutTracker; private volatile long cacheEvictionTimeThresholdNanos; private final MetadataStore metadataStore; @@ -245,6 +246,7 @@ private ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore, this.config = config; this.mbean = new ManagedLedgerFactoryMBeanImpl(this); this.entryCacheManager = new RangeEntryCacheManagerImpl(this, scheduledExecutor, openTelemetry); + this.readEntryTimeoutTracker = new ReadEntryTimeoutTracker(scheduledExecutor); this.statsTask = scheduledExecutor.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::refreshStats), 0, config.getStatsPeriodSeconds(), TimeUnit.SECONDS); this.flushCursorsTask = scheduledExecutor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::flushCursors), @@ -324,6 +326,10 @@ public synchronized void doCacheEviction() { entryCacheManager.doCacheEviction(); } + ReadEntryTimeoutTracker getReadEntryTimeoutTracker() { + return readEntryTimeoutTracker; + } + /** * Waits for all pending cache evictions based on total cache size or entry TTL to complete. * This is for testing purposes only, so that we can ensure all cache evictions are done before proceeding with @@ -662,6 +668,7 @@ public CompletableFuture shutdownAsync() throws ManagedLedgerException { statsTask.cancel(true); flushCursorsTask.cancel(true); + readEntryTimeoutTracker.close(); cacheEvictionExecutor.shutdownNow(); List ledgerNames = new ArrayList<>(this.ledgers.keySet()); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 01d5d20065337..74f5ff0b3b01b 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -56,7 +56,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; @@ -349,9 +348,6 @@ public boolean isFenced() { .newUpdater(ManagedLedgerImpl.class, "addOpCount"); private volatile long addOpCount = 0; - // Pending reads ordered by timeout so checks can stop at the first operation that has not expired. - private final PriorityBlockingQueue pendingReadCallbacks = new PriorityBlockingQueue<>(); - /** * Queue of pending entries to be added to the managed ledger. Typically, entries are queued when a new ledger is. * created asynchronously and hence there is no ready ledger to write into. @@ -527,7 +523,7 @@ public void operationFailed(MetaStoreException e) { } }); - scheduleTimeoutTask(); + scheduleAddEntryTimeoutTask(); } protected ManagedLedgerInterceptor.LastEntryHandle createLastEntryHandle(LedgerHandle lh) { @@ -2435,13 +2431,10 @@ protected void asyncReadEntry(ReadHandle ledger, Position position, ReadEntryCal if (config.getReadEntryTimeoutSeconds() > 0) { // set readOpCount to uniquely validate if ReadEntryCallbackWrapper is already recycled long readOpCount = READ_OP_COUNT_UPDATER.incrementAndGet(this); - ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, position.getLedgerId(), + ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(this, position.getLedgerId(), position.getEntryId(), callback, readOpCount, ctx); - PendingReadCallback pendingReadCallback = new PendingReadCallback(readOpCount, - timeoutAtNanos(config.getReadEntryTimeoutSeconds()), position.getLedgerId(), position.getEntryId(), - readCallback); - readCallback.pendingReadCallback = pendingReadCallback; - pendingReadCallbacks.add(pendingReadCallback); + readCallback.readTimeout = factory.getReadEntryTimeoutTracker().add(readCallback, readOpCount, + timeoutAtNanos(config.getReadEntryTimeoutSeconds())); entryCache.asyncReadEntry(ledger, position, readCallback, readOpCount); } else { entryCache.asyncReadEntry(ledger, position, callback, ctx); @@ -2454,12 +2447,10 @@ protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry if (config.getReadEntryTimeoutSeconds() > 0) { // set readOpCount to uniquely validate if ReadEntryCallbackWrapper is already recycled long readOpCount = READ_OP_COUNT_UPDATER.incrementAndGet(this); - ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(name, ledger.getId(), firstEntry, + ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(this, ledger.getId(), firstEntry, opReadEntry, readOpCount, ctx); - PendingReadCallback pendingReadCallback = new PendingReadCallback(readOpCount, - timeoutAtNanos(config.getReadEntryTimeoutSeconds()), ledger.getId(), firstEntry, readCallback); - readCallback.pendingReadCallback = pendingReadCallback; - pendingReadCallbacks.add(pendingReadCallback); + readCallback.readTimeout = factory.getReadEntryTimeoutTracker().add(readCallback, readOpCount, + timeoutAtNanos(config.getReadEntryTimeoutSeconds())); entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, readCallback, readOpCount); } else { entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, opReadEntry, ctx); @@ -2470,53 +2461,18 @@ private static long timeoutAtNanos(long timeoutSec) { return System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSec); } - static final class PendingReadCallback implements Comparable { - private static final AtomicReferenceFieldUpdater - CALLBACK_UPDATER = AtomicReferenceFieldUpdater - .newUpdater(PendingReadCallback.class, ReadEntryCallbackWrapper.class, "callback"); - - final long readOpCount; - final long timeoutAtNanos; - final long ledgerId; - final long entryId; - volatile ReadEntryCallbackWrapper callback; - - PendingReadCallback(long readOpCount, long timeoutAtNanos, long ledgerId, long entryId, - ReadEntryCallbackWrapper callback) { - this.readOpCount = readOpCount; - this.timeoutAtNanos = timeoutAtNanos; - this.ledgerId = ledgerId; - this.entryId = entryId; - this.callback = callback; - } - - @Override - public int compareTo(PendingReadCallback other) { - int result = Long.compare(timeoutAtNanos, other.timeoutAtNanos); - return result != 0 ? result : Long.compare(readOpCount, other.readOpCount); - } - - ReadEntryCallbackWrapper clearCallback() { - return CALLBACK_UPDATER.getAndSet(this, null); - } - - void clearCallback(ReadEntryCallbackWrapper callback) { - CALLBACK_UPDATER.compareAndSet(this, callback, null); - } - } - static final class ReadEntryCallbackWrapper implements ReadEntryCallback, ReadEntriesCallback { volatile ReadEntryCallback readEntryCallback; volatile ReadEntriesCallback readEntriesCallback; - String name; + volatile ManagedLedgerImpl managedLedger; long ledgerId; long entryId; volatile long readOpCount = -1; private static final AtomicLongFieldUpdater READ_OP_COUNT_UPDATER = AtomicLongFieldUpdater.newUpdater(ReadEntryCallbackWrapper.class, "readOpCount"); volatile Object cntx; - volatile PendingReadCallback pendingReadCallback; + volatile ReadEntryTimeoutTracker.ReadTimeoutWrapper readTimeout; final Handle recyclerHandle; @@ -2524,10 +2480,10 @@ private ReadEntryCallbackWrapper(Handle recyclerHandle this.recyclerHandle = recyclerHandle; } - static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntryCallback callback, - long readOpCount, Object ctx) { + static ReadEntryCallbackWrapper create(ManagedLedgerImpl managedLedger, long ledgerId, long entryId, + ReadEntryCallback callback, long readOpCount, Object ctx) { ReadEntryCallbackWrapper readCallback = RECYCLER.get(); - readCallback.name = name; + readCallback.managedLedger = managedLedger; readCallback.ledgerId = ledgerId; readCallback.entryId = entryId; readCallback.readEntryCallback = callback; @@ -2536,10 +2492,10 @@ static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, return readCallback; } - static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntriesCallback callback, - long readOpCount, Object ctx) { + static ReadEntryCallbackWrapper create(ManagedLedgerImpl managedLedger, long ledgerId, long entryId, + ReadEntriesCallback callback, long readOpCount, Object ctx) { ReadEntryCallbackWrapper readCallback = RECYCLER.get(); - readCallback.name = name; + readCallback.managedLedger = managedLedger; readCallback.ledgerId = ledgerId; readCallback.entryId = entryId; readCallback.readEntriesCallback = callback; @@ -2602,6 +2558,15 @@ private long reOpCount(Object ctx) { return (ctx instanceof Long) ? (long) ctx : -1; } + boolean shouldTriggerReadTimeout() { + ManagedLedgerImpl ledger = managedLedger; + if (ledger == null) { + return false; + } + State state = STATE_UPDATER.get(ledger); + return state != State.Closed && !state.isFenced(); + } + public void readFailed(ManagedLedgerException exception, Object ctx) { if (readEntryCallback != null) { readEntryFailed(exception, ctx); @@ -2615,17 +2580,17 @@ public void readFailed(ManagedLedgerException exception, Object ctx) { private boolean recycle(long readOpCount) { if (readOpCount != -1 && READ_OP_COUNT_UPDATER.compareAndSet(ReadEntryCallbackWrapper.this, readOpCount, -1)) { - PendingReadCallback pendingCallback = pendingReadCallback; - if (pendingCallback != null) { - pendingCallback.clearCallback(this); + ReadEntryTimeoutTracker.ReadTimeoutWrapper timeout = readTimeout; + if (timeout != null) { + timeout.clearCallback(this); } readEntryCallback = null; readEntriesCallback = null; ledgerId = -1; entryId = -1; - name = null; + managedLedger = null; cntx = null; - pendingReadCallback = null; + readTimeout = null; recyclerHandle.recycle(this); return true; } @@ -4769,29 +4734,20 @@ protected boolean checkAndCompleteLedgerOpTask(int rc, LedgerHandle lh, Object c return false; } - private void scheduleTimeoutTask() { - // disable timeout task checker if timeout <= 0 - if (config.getAddEntryTimeoutSeconds() > 0 || config.getReadEntryTimeoutSeconds() > 0) { - long timeoutSec = Math.min(config.getAddEntryTimeoutSeconds(), config.getReadEntryTimeoutSeconds()); - timeoutSec = timeoutSec <= 0 - ? Math.max(config.getAddEntryTimeoutSeconds(), config.getReadEntryTimeoutSeconds()) - : timeoutSec; + private void scheduleAddEntryTimeoutTask() { + if (config.getAddEntryTimeoutSeconds() > 0) { + long timeoutSec = config.getAddEntryTimeoutSeconds(); this.timeoutTask = this.scheduledExecutor.scheduleAtFixedRate( - this::checkTimeouts, timeoutSec, timeoutSec, TimeUnit.SECONDS); + this::checkAddTimeout, timeoutSec, timeoutSec, TimeUnit.SECONDS); } } - private void checkTimeouts() { + private void checkAddTimeout() { final State state = STATE_UPDATER.get(this); if (state == State.Closed || state.isFenced()) { return; } - checkAddTimeout(); - checkReadTimeout(); - } - - private void checkAddTimeout() { long timeoutSec = config.getAddEntryTimeoutSeconds(); if (timeoutSec < 1) { return; @@ -4811,33 +4767,6 @@ private void checkAddTimeout() { } } - private void checkReadTimeout() { - long timeoutSec = config.getReadEntryTimeoutSeconds(); - if (timeoutSec < 1) { - return; - } - long now = System.nanoTime(); - while (true) { - PendingReadCallback pendingCallback = pendingReadCallbacks.peek(); - if (pendingCallback == null || pendingCallback.timeoutAtNanos > now) { - break; - } - pendingCallback = pendingReadCallbacks.poll(); - if (pendingCallback == null) { - break; - } - ReadEntryCallbackWrapper callback = pendingCallback.clearCallback(); - if (callback != null) { - log.warn().attr("ledgerId", pendingCallback.ledgerId) - .attr("entryId", pendingCallback.entryId) - .attr("timeoutSec", timeoutSec) - .log("Read entry timeout"); - callback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException), - pendingCallback.readOpCount); - } - } - } - @Override public long getOffloadedSize() { long offloadedSize = 0; diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadEntryTimeoutTracker.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadEntryTimeoutTracker.java new file mode 100644 index 0000000000000..49a858c0848c7 --- /dev/null +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadEntryTimeoutTracker.java @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.bookkeeper.mledger.impl; + +import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.createManagedLedgerException; +import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables; +import com.google.common.annotations.VisibleForTesting; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import lombok.CustomLog; +import org.apache.bookkeeper.client.BKException; +import org.apache.bookkeeper.common.util.OrderedScheduler; +import org.jctools.queues.MpscUnboundedArrayQueue; + +@CustomLog +class ReadEntryTimeoutTracker implements AutoCloseable { + private static final int READ_TIMEOUT_QUEUE_CHUNK_SIZE = 128 * 1024; + private static final int CHECK_INTERVAL_SECONDS = 1; + + private final MpscUnboundedArrayQueue timeoutQueue = new MpscUnboundedArrayQueue<>( + READ_TIMEOUT_QUEUE_CHUNK_SIZE); + private final AtomicInteger timeoutQueueSize = new AtomicInteger(); + private final ScheduledFuture timeoutTask; + + ReadEntryTimeoutTracker(OrderedScheduler scheduledExecutor) { + this.timeoutTask = scheduledExecutor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::checkTimeouts), + CHECK_INTERVAL_SECONDS, CHECK_INTERVAL_SECONDS, TimeUnit.SECONDS); + } + + ReadTimeoutWrapper add(ManagedLedgerImpl.ReadEntryCallbackWrapper callback, long readOpCount, + long timeoutAtNanos) { + ReadTimeoutWrapper timeout = new ReadTimeoutWrapper(readOpCount, timeoutAtNanos, callback); + timeoutQueue.offer(timeout); + timeoutQueueSize.incrementAndGet(); + return timeout; + } + + @VisibleForTesting + synchronized void checkTimeouts() { + long now = System.nanoTime(); + int entriesToProcess = timeoutQueueSize.get(); + for (int i = 0; i < entriesToProcess; i++) { + ReadTimeoutWrapper timeout = timeoutQueue.poll(); + if (timeout == null) { + return; + } + timeoutQueueSize.decrementAndGet(); + ManagedLedgerImpl.ReadEntryCallbackWrapper callback = timeout.getCallback(); + if (callback == null) { + continue; + } + if (!callback.shouldTriggerReadTimeout()) { + timeout.clearCallback(callback); + continue; + } + if (timeout.timeoutAtNanos > now) { + requeue(timeout); + continue; + } + callback = timeout.clearCallback(); + if (callback != null) { + log.warn() + .attr("overdueNanos", now - timeout.timeoutAtNanos) + .log("Read entry timeout"); + callback.readFailed(createManagedLedgerException(BKException.Code.TimeoutException), + timeout.readOpCount); + } + } + } + + private void requeue(ReadTimeoutWrapper timeout) { + timeoutQueue.offer(timeout); + timeoutQueueSize.incrementAndGet(); + } + + @Override + public void close() { + timeoutTask.cancel(false); + } + + @VisibleForTesting + int pendingTimeoutCount() { + return timeoutQueueSize.get(); + } + + static final class ReadTimeoutWrapper { + private static final AtomicReferenceFieldUpdater CALLBACK_UPDATER = AtomicReferenceFieldUpdater + .newUpdater(ReadTimeoutWrapper.class, ManagedLedgerImpl.ReadEntryCallbackWrapper.class, "callback"); + + final long readOpCount; + final long timeoutAtNanos; + volatile ManagedLedgerImpl.ReadEntryCallbackWrapper callback; + + ReadTimeoutWrapper(long readOpCount, long timeoutAtNanos, + ManagedLedgerImpl.ReadEntryCallbackWrapper callback) { + this.readOpCount = readOpCount; + this.timeoutAtNanos = timeoutAtNanos; + this.callback = callback; + } + + ManagedLedgerImpl.ReadEntryCallbackWrapper getCallback() { + return callback; + } + + ManagedLedgerImpl.ReadEntryCallbackWrapper clearCallback() { + return CALLBACK_UPDATER.getAndSet(this, null); + } + + void clearCallback(ManagedLedgerImpl.ReadEntryCallbackWrapper callback) { + CALLBACK_UPDATER.compareAndSet(this, callback, null); + } + } +} diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index b6979002bc655..110436011c436 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -3489,6 +3489,33 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) { ledger.close(); } + @Test + public void testCompletedReadEntryTimeoutsAreRemovedFromSharedTracker() throws Exception { + ManagedLedgerConfig config = initManagedLedgerConfig(new ManagedLedgerConfig()).setReadEntryTimeoutSeconds(60); + ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("completed_read_timeout_tracker_test", config); + Position position = ledger.addEntry("entry-1".getBytes()); + + CompletableFuture readComplete = new CompletableFuture<>(); + ledger.asyncReadEntry(position, new ReadEntryCallback() { + @Override + public void readEntryComplete(Entry entry, Object ctx) { + entry.release(); + readComplete.complete(null); + } + + @Override + public void readEntryFailed(ManagedLedgerException exception, Object ctx) { + readComplete.completeExceptionally(exception); + } + }, null); + + readComplete.get(5, TimeUnit.SECONDS); + factory.getReadEntryTimeoutTracker().checkTimeouts(); + assertEquals(factory.getReadEntryTimeoutTracker().pendingTimeoutCount(), 0); + + ledger.close(); + } + @Test public void testAddEntryResponseTimeout() throws Exception { // Create ML with feature Add Entry Timeout Check.