You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2026/09/07 08:19:33 UTC
(pulsar) branch master updated: [fix][ml] Track all pending read callbacks for timeouts (#26081)
This is an automated email from the ASF dual-hosted git repository.
Technoboy- pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 9ba61bd95de7 [fix][ml] Track all pending read callbacks for timeouts (#26081)
9ba61bd95de7 is described below
commit 9ba61bd95de78ab7891da5e0f4b9174acce0c9ff
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Mon Sep 7 16:19:24 2026 +0800
[fix][ml] Track all pending read callbacks for timeouts (#26081)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 7 +
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 235 ++++++++-------------
.../mledger/impl/ReadEntryTimeoutTracker.java | 116 ++++++++++
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 177 ++++++++++++++++
4 files changed, 388 insertions(+), 147 deletions(-)
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 e855a21c17d9..0953980c534e 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
@@ -135,6 +135,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;
@@ -247,6 +248,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
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),
@@ -326,6 +328,10 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
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
@@ -664,6 +670,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
statsTask.cancel(true);
flushCursorsTask.cancel(true);
+ readEntryTimeoutTracker.close();
cacheEvictionExecutor.shutdownNow();
List<String> 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 bd59826c3ea1..7740bca174bf 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
@@ -30,8 +30,6 @@ import com.google.common.collect.Range;
import io.github.merlimat.slog.Logger;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
-import io.netty.util.Recycler;
-import io.netty.util.Recycler.Handle;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Collection;
@@ -62,6 +60,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
@@ -341,19 +340,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
protected final ManagedLedgerMBeanImpl mbean;
protected final Clock clock;
- private static final AtomicLongFieldUpdater<ManagedLedgerImpl> READ_OP_COUNT_UPDATER = AtomicLongFieldUpdater
- .newUpdater(ManagedLedgerImpl.class, "readOpCount");
- private volatile long readOpCount = 0;
protected static final AtomicLongFieldUpdater<ManagedLedgerImpl> ADD_OP_COUNT_UPDATER = AtomicLongFieldUpdater
.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<ManagedLedgerImpl, ReadEntryCallbackWrapper>
- LAST_READ_CALLBACK_UPDATER = AtomicReferenceFieldUpdater
- .newUpdater(ManagedLedgerImpl.class, ReadEntryCallbackWrapper.class, "lastReadCallback");
-
/**
* 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.
@@ -529,7 +519,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
});
- scheduleTimeoutTask();
+ scheduleAddEntryTimeoutTask();
}
protected ManagedLedgerInterceptor.LastEntryHandle createLastEntryHandle(LedgerHandle lh) {
@@ -2446,13 +2436,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
protected void asyncReadEntry(ReadHandle ledger, Position position, ReadEntryCallback callback, Object ctx) {
mbean.addEntriesRead(1);
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);
- lastReadCallback = readCallback;
- entryCache.asyncReadEntry(ledger, position, readCallback, readOpCount);
+ ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(this, position.getLedgerId(),
+ position.getEntryId(), callback, ctx, timeoutAtNanos(config.getReadEntryTimeoutSeconds()));
+ entryCache.asyncReadEntry(ledger, position, readCallback, ctx);
+ if (readCallback.registerTimeout()) {
+ factory.getReadEntryTimeoutTracker().add(readCallback);
+ }
} else {
entryCache.asyncReadEntry(ledger, position, callback, ctx);
}
@@ -2462,70 +2451,66 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
Object ctx) {
IntSupplier expectedReadCount = opReadEntry.cursor::getNumberOfCursorsAtSamePositionOrBefore;
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);
- lastReadCallback = readCallback;
- entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, readCallback, readOpCount);
+ ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(this, ledger.getId(), firstEntry,
+ opReadEntry, ctx, timeoutAtNanos(config.getReadEntryTimeoutSeconds()));
+ entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, readCallback, ctx);
+ if (readCallback.registerTimeout()) {
+ factory.getReadEntryTimeoutTracker().add(readCallback);
+ }
} 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 ReadEntryCallbackWrapper implements ReadEntryCallback, ReadEntriesCallback {
- volatile ReadEntryCallback readEntryCallback;
- volatile ReadEntriesCallback readEntriesCallback;
- String name;
- long ledgerId;
- long entryId;
- volatile long readOpCount = -1;
- private static final AtomicLongFieldUpdater<ReadEntryCallbackWrapper> READ_OP_COUNT_UPDATER =
- AtomicLongFieldUpdater.newUpdater(ReadEntryCallbackWrapper.class, "readOpCount");
- volatile long createdTime = -1;
- volatile Object cntx;
-
- final Handle<ReadEntryCallbackWrapper> recyclerHandle;
-
- private ReadEntryCallbackWrapper(Handle<ReadEntryCallbackWrapper> recyclerHandle) {
- this.recyclerHandle = recyclerHandle;
- }
-
- static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntryCallback callback,
- long readOpCount, long createdTime, Object ctx) {
- ReadEntryCallbackWrapper readCallback = RECYCLER.get();
- readCallback.name = name;
- readCallback.ledgerId = ledgerId;
- readCallback.entryId = entryId;
- readCallback.readEntryCallback = callback;
- readCallback.cntx = ctx;
- readCallback.readOpCount = readOpCount;
- readCallback.createdTime = createdTime;
- return readCallback;
- }
-
- static ReadEntryCallbackWrapper create(String name, long ledgerId, long entryId, ReadEntriesCallback callback,
- long readOpCount, long createdTime, Object ctx) {
- ReadEntryCallbackWrapper readCallback = RECYCLER.get();
- readCallback.name = name;
- readCallback.ledgerId = ledgerId;
- readCallback.entryId = entryId;
- readCallback.readEntriesCallback = callback;
- readCallback.cntx = ctx;
- readCallback.readOpCount = readOpCount;
- readCallback.createdTime = createdTime;
- return readCallback;
+ private static final int WAITING = 0;
+ private static final int TRACKED = 1;
+ private static final int COMPLETED = 2;
+ private static final AtomicIntegerFieldUpdater<ReadEntryCallbackWrapper> STATE_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(ReadEntryCallbackWrapper.class, "completionState");
+
+ final ReadEntryCallback readEntryCallback;
+ final ReadEntriesCallback readEntriesCallback;
+ final ManagedLedgerImpl managedLedger;
+ final String managedLedgerName;
+ final long ledgerId;
+ final long entryId;
+ final long timeoutAtNanos;
+ final Object cntx;
+ private volatile int completionState;
+
+ private ReadEntryCallbackWrapper(ManagedLedgerImpl managedLedger, long ledgerId, long entryId,
+ ReadEntryCallback readEntryCallback, ReadEntriesCallback readEntriesCallback,
+ Object ctx, long timeoutAtNanos) {
+ this.managedLedger = managedLedger;
+ this.managedLedgerName = managedLedger.name;
+ this.ledgerId = ledgerId;
+ this.entryId = entryId;
+ this.readEntryCallback = readEntryCallback;
+ this.readEntriesCallback = readEntriesCallback;
+ this.cntx = ctx;
+ this.timeoutAtNanos = timeoutAtNanos;
+ }
+
+ static ReadEntryCallbackWrapper create(ManagedLedgerImpl managedLedger, long ledgerId, long entryId,
+ ReadEntryCallback callback, Object ctx, long timeoutAtNanos) {
+ return new ReadEntryCallbackWrapper(managedLedger, ledgerId, entryId, callback, null, ctx, timeoutAtNanos);
+ }
+
+ static ReadEntryCallbackWrapper create(ManagedLedgerImpl managedLedger, long ledgerId, long entryId,
+ ReadEntriesCallback callback, Object ctx, long timeoutAtNanos) {
+ return new ReadEntryCallbackWrapper(managedLedger, ledgerId, entryId, null, callback, ctx, timeoutAtNanos);
}
@Override
public void readEntryComplete(Entry entry, Object ctx) {
- long reOpCount = reOpCount(ctx);
- ReadEntryCallback callback = this.readEntryCallback;
- Object cbCtx = this.cntx;
- if (recycle(reOpCount)) {
- callback.readEntryComplete(entry, cbCtx);
+ if (complete()) {
+ readEntryCallback.readEntryComplete(entry, cntx);
} else {
slog.debug().attr("ledgerId", ledgerId).attr("entryId", entryId).log("Read entry already completed");
entry.release();
@@ -2534,11 +2519,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
- long reOpCount = reOpCount(ctx);
- ReadEntryCallback callback = this.readEntryCallback;
- Object cbCtx = this.cntx;
- if (recycle(reOpCount)) {
- callback.readEntryFailed(exception, cbCtx);
+ if (complete()) {
+ readEntryCallback.readEntryFailed(exception, cntx);
} else {
slog.debug().attr("ledgerId", ledgerId).attr("entryId", entryId).log("Read entry already completed");
}
@@ -2546,11 +2528,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public void readEntriesComplete(List<Entry> returnedEntries, Object ctx) {
- long reOpCount = reOpCount(ctx);
- ReadEntriesCallback callback = this.readEntriesCallback;
- Object cbCtx = this.cntx;
- if (recycle(reOpCount)) {
- callback.readEntriesComplete(returnedEntries, cbCtx);
+ if (complete()) {
+ readEntriesCallback.readEntriesComplete(returnedEntries, cntx);
} else {
slog.debug().attr("ledgerId", ledgerId).attr("entryId", entryId).log("Read entry already completed");
returnedEntries.forEach(Entry::release);
@@ -2559,51 +2538,40 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
- long reOpCount = reOpCount(ctx);
- ReadEntriesCallback callback = this.readEntriesCallback;
- Object cbCtx = this.cntx;
- if (recycle(reOpCount)) {
- callback.readEntriesFailed(exception, cbCtx);
+ if (complete()) {
+ readEntriesCallback.readEntriesFailed(exception, cntx);
} else {
slog.debug().attr("ledgerId", ledgerId).attr("entryId", entryId).log("Read entry already completed");
}
}
- private long reOpCount(Object ctx) {
- return (ctx instanceof Long) ? (long) ctx : -1;
+ boolean isCompleted() {
+ return completionState == COMPLETED;
+ }
+
+ boolean registerTimeout() {
+ return STATE_UPDATER.compareAndSet(this, WAITING, TRACKED);
}
- public void readFailed(ManagedLedgerException exception, Object ctx) {
+ boolean triggerReadTimeout(ManagedLedgerException exception) {
+ if (!STATE_UPDATER.compareAndSet(this, TRACKED, COMPLETED)) {
+ return false;
+ }
+ managedLedger.getExecutor().execute(() -> notifyReadFailed(exception));
+ return true;
+ }
+
+ private void notifyReadFailed(ManagedLedgerException exception) {
if (readEntryCallback != null) {
- readEntryFailed(exception, ctx);
+ readEntryCallback.readEntryFailed(exception, cntx);
} else if (readEntriesCallback != null) {
- readEntriesFailed(exception, ctx);
- }
- // It happens when timeout-thread and read-callback both recycles at the same time.
- // this read-callback has already been recycled so, do nothing..
- }
-
- private boolean recycle(long readOpCount) {
- if (readOpCount != -1
- && READ_OP_COUNT_UPDATER.compareAndSet(ReadEntryCallbackWrapper.this, readOpCount, -1)) {
- createdTime = -1;
- readEntryCallback = null;
- readEntriesCallback = null;
- ledgerId = -1;
- entryId = -1;
- name = null;
- recyclerHandle.recycle(this);
- return true;
+ readEntriesCallback.readEntriesFailed(exception, cntx);
}
- return false;
}
- private static final Recycler<ReadEntryCallbackWrapper> RECYCLER = new Recycler<ReadEntryCallbackWrapper>() {
- @Override
- protected ReadEntryCallbackWrapper newObject(Handle<ReadEntryCallbackWrapper> handle) {
- return new ReadEntryCallbackWrapper(handle);
- }
- };
+ private boolean complete() {
+ return STATE_UPDATER.getAndSet(this, COMPLETED) != COMPLETED;
+ }
}
@@ -4764,29 +4732,20 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
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;
@@ -4806,24 +4765,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
}
- private void checkReadTimeout() {
- long timeoutSec = config.getReadEntryTimeoutSeconds();
- 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);
- }
- }
-
@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 000000000000..6bb169520e80
--- /dev/null
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadEntryTimeoutTracker.java
@@ -0,0 +1,116 @@
+/*
+ * 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 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<ManagedLedgerImpl.ReadEntryCallbackWrapper> 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);
+ }
+
+ void add(ManagedLedgerImpl.ReadEntryCallbackWrapper callback) {
+ timeoutQueue.offer(callback);
+ timeoutQueueSize.incrementAndGet();
+ }
+
+ @VisibleForTesting
+ synchronized void checkTimeouts() {
+ long now = System.nanoTime();
+ // This is intentionally O(all tracked reads). Read-entry timeout is disabled by default, the
+ // tracker runs once per second, and the MPSC queue keeps this maintenance path cheap while
+ // avoiding per-ledger timeout tasks. The assumption is that deployments enabling this feature
+ // prefer one lightweight broker-level scan over creating a scheduled task per ManagedLedger or
+ // per read. Expired callbacks are handed back to the owning ledger's executor, so this shared
+ // scan does not serialize completion callbacks across unrelated ledgers.
+ int entriesToProcess = timeoutQueueSize.get();
+ int timeoutCount = 0;
+ long maxOverdueNanos = 0;
+ String sampleLedgerName = null;
+ long sampleLedgerId = -1;
+ long sampleEntryId = -1;
+ for (int i = 0; i < entriesToProcess; i++) {
+ ManagedLedgerImpl.ReadEntryCallbackWrapper callback = timeoutQueue.poll();
+ if (callback == null) {
+ break;
+ }
+ timeoutQueueSize.decrementAndGet();
+ if (callback.isCompleted()) {
+ continue;
+ }
+ if (callback.timeoutAtNanos > now) {
+ requeue(callback);
+ continue;
+ }
+ if (callback.triggerReadTimeout(createManagedLedgerException(BKException.Code.TimeoutException))) {
+ timeoutCount++;
+ long overdueNanos = now - callback.timeoutAtNanos;
+ maxOverdueNanos = Math.max(maxOverdueNanos, overdueNanos);
+ if (sampleLedgerName == null) {
+ sampleLedgerName = callback.managedLedgerName;
+ sampleLedgerId = callback.ledgerId;
+ sampleEntryId = callback.entryId;
+ }
+ }
+ }
+ if (timeoutCount > 0) {
+ log.warn()
+ .attr("timeoutCount", timeoutCount)
+ .attr("sampleLedgerName", sampleLedgerName)
+ .attr("sampleLedgerId", sampleLedgerId)
+ .attr("sampleEntryId", sampleEntryId)
+ .attr("maxOverdueNanos", maxOverdueNanos)
+ .log("Read entry timeouts");
+ }
+ }
+
+ private void requeue(ManagedLedgerImpl.ReadEntryCallbackWrapper callback) {
+ timeoutQueue.offer(callback);
+ timeoutQueueSize.incrementAndGet();
+ }
+
+ @Override
+ public void close() {
+ timeoutTask.cancel(false);
+ }
+
+ @VisibleForTesting
+ int pendingTimeoutCount() {
+ return timeoutQueueSize.get();
+ }
+}
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 2a1b03b1591e..0cad4466b3f9 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
@@ -3433,6 +3433,183 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
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<LedgerEntries> interceptReadAsync(long ledgerId, long firstEntry, long lastEntry,
+ LedgerEntries entries) {
+ return CompletableFuture.supplyAsync(() -> entries,
+ CompletableFuture.delayedExecutor(3, TimeUnit.SECONDS));
+ }
+ });
+
+ AtomicReference<ManagedLedgerException> responseException1 = new AtomicReference<>();
+ AtomicReference<ManagedLedgerException> 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 testReadEntryTimeoutCallbackRunsOnManagedLedgerExecutor() throws Exception {
+ ManagedLedgerConfig config = initManagedLedgerConfig(new ManagedLedgerConfig()).setReadEntryTimeoutSeconds(1);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("timeout_executor_test", config);
+ Position position = ledger.addEntry("entry-1".getBytes());
+
+ factory.getEntryCacheManager().clear();
+
+ bkc.setReadHandleInterceptor(new PulsarMockReadHandleInterceptor() {
+ @Override
+ public CompletableFuture<LedgerEntries> interceptReadAsync(long ledgerId, long firstEntry, long lastEntry,
+ LedgerEntries entries) {
+ return CompletableFuture.supplyAsync(() -> entries,
+ CompletableFuture.delayedExecutor(3, TimeUnit.SECONDS));
+ }
+ });
+
+ CountDownLatch executorBlocked = new CountDownLatch(1);
+ CountDownLatch releaseExecutor = new CountDownLatch(1);
+ AtomicReference<ManagedLedgerException> responseException = new AtomicReference<>();
+ try {
+ ledger.getExecutor().execute(() -> {
+ executorBlocked.countDown();
+ try {
+ releaseExecutor.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue(executorBlocked.await(5, TimeUnit.SECONDS));
+
+ ledger.asyncReadEntry(position, new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ entry.release();
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ responseException.set(exception);
+ }
+ }, null);
+
+ Awaitility.await().untilAsserted(() -> {
+ factory.getReadEntryTimeoutTracker().checkTimeouts();
+ assertEquals(factory.getReadEntryTimeoutTracker().pendingTimeoutCount(), 0);
+ });
+ assertNull(responseException.get());
+
+ releaseExecutor.countDown();
+ Awaitility.await().untilAsserted(() -> {
+ assertNotNull(responseException.get());
+ assertTrue(responseException.get().getMessage()
+ .startsWith(BKException.getMessage(BKException.Code.TimeoutException)));
+ });
+ } finally {
+ releaseExecutor.countDown();
+ 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<Void> 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 testCompletedReadEntryIsNotRegisteredForReadTimeout() throws Exception {
+ Object expectedCtx = new Object();
+ CompletableFuture<Object> callbackContext = new CompletableFuture<>();
+ ReadEntryCallback callback = new ReadEntryCallback() {
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ callbackContext.complete(ctx);
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ callbackContext.complete(ctx);
+ }
+ };
+ ManagedLedgerImpl.ReadEntryCallbackWrapper readCallback = ManagedLedgerImpl.ReadEntryCallbackWrapper.create(
+ mock(ManagedLedgerImpl.class), 1L, 2L, callback, expectedCtx,
+ System.nanoTime() + TimeUnit.SECONDS.toNanos(60));
+
+ readCallback.readEntryFailed(new ManagedLedgerException("completed"), null);
+
+ assertSame(callbackContext.get(5, TimeUnit.SECONDS), expectedCtx);
+ assertTrue(readCallback.isCompleted());
+ assertFalse(readCallback.registerTimeout());
+ assertFalse(readCallback.triggerReadTimeout(new ManagedLedgerException("timeout")));
+ }
+
@Test
public void testAddEntryResponseTimeout() throws Exception {
// Create ML with feature Add Entry Timeout Check.