Skip to content

[fix][txn] End pending ack replay on read failures it does not classify #26380

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
lhotari wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

[fix][txn] End pending ack replay on read failures it does not classify #26380

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5da5f22
[fix][txn] Stop the pending ack replay loop from spinning forever
lhotari Aug 18, 2026
52f030b
[fix][txn] End pending ack replay on read failures it does not classify
lhotari Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
223 changes: 202 additions & 21 deletions ...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -405,7 +405,7 @@ class PendingAckReplay implements Runnable {
public void run() {
try {
if (cursor.isClosed()) {
pendingAckReplyCallBack.replayFailed(new ManagedLedgerException
failReplay(new ManagedLedgerException
.CursorAlreadyClosedException("MLPendingAckStore cursor have been closed."));
log.warn("MLPendingAckStore cursor have been closed, close replay thread");
return;
Expand Down Expand Up @@ -434,24 +434,92 @@ public void run() {
entry.release();
clearUselessLogData();
} else {
// Covers a read that was issued but whose callback never arrives: no failure is
// ever delivered, so readEntriesFailed cannot notice the close. A cursor closed
// between two reads takes a different path -- that read fails synchronously in
// ManagedCursorImpl#asyncReadEntriesWithSkip, which clears isReadable and ends
// the loop through the existing read failure handling.
if (cursor.isClosed()) {
failReplay(new ManagedLedgerException
.CursorAlreadyClosedException("MLPendingAckStore cursor was closed "
+ "while replaying."));
log.warn("MLPendingAckStore cursor was closed while replaying, close replay thread");
return;
}
try {
Thread.sleep(1);
} catch (InterruptedException e) {
if (Thread.interrupted()) {
log.error()
.exception(e)
.log("Transaction pending replay thread interrupt!");
}
// Restore the interrupt flag and stop. The replay is incomplete, so it must not
// be reported as successful: replayFailed() lets PendingAckHandleImpl decide
// whether to retry instead of leaving the handle Ready with partial state.
Thread.currentThread().interrupt();
failReplay(e);
log.warn()
.exception(e)
.log("Transaction pending ack replay thread was interrupted");
return;
}
}
}
} catch (Exception e) {
pendingAckReplyCallBack.replayFailed(e);
failReplay(e);
log.error().exception(e).log("Pending ack recover fail");
return;
}
stopReplay();
// Written by readEntriesFailed before the volatile isReadable write that made fillQueue()
// return false, so reading it here is safely ordered.
ManagedLedgerException attemptFailure = fillEntryQueueCallback.replayAttemptFailure();
if (attemptFailure != null && lastConfirmedEntry.compareTo(currentLoadPosition) > 0) {
// A read failed before the replay reached the last confirmed entry, so the replayed
// pending ack state is incomplete: report the attempt as failed and let
// PendingAckHandleImpl#exceptionHandleFuture choose between a backoff-paced retry and
// failing the subscription fast. If the loop had instead already caught up, the state
// is complete, and a failure of a still-in-flight read beyond it must not fail the
// attempt.
//
// Reads run ahead of processing: entries this attempt read but never processed have
// already advanced the shared cursor's read position, and stopReplay() has just
// released them. The cursor is cached by the managed ledger, so the next attempt would
// resume at that advanced position and silently skip them, then report the incomplete
// replay as successful. Rewinding puts the read position back to the first entry after
// the mark-delete position, which is exactly where the next attempt's
// currentLoadPosition starts. Re-reading entries this attempt already applied is safe:
// the replay handlers deduplicate against the state they have already built.
// No read is in flight here -- the recorded failure was the only outstanding one, and
// fillQueue() has issued none since -- so nothing can move the read position after this.
cursor.rewind();
failReplay(attemptFailure);
return;
}
pendingAckReplyCallBack.replayComplete();
}

/**
* Ends the replay. Entries that were read but never processed are released, and any read that is
* still in flight will release its entries itself instead of queueing them, so nothing is left
* holding a buffer once the replay thread is gone. Called on every exit from {@link #run()}.
*/
private void stopReplay() {
for (Entry entry : fillEntryQueueCallback.stopAndDrain()) {
entry.release();
}
}

/**
* Ends this replay attempt as failed. The store is abandoned afterwards: PendingAckHandleImpl
* either retries with a brand new store or fails the handle, and in both cases
* pendingAckStoreFuture stops referring to this one. Closing the buffered writer here cancels
* its recurring flush timer, which nothing else would ever reach. That touches no shared state
* -- the cursor and the managed ledger belong to the managed ledger factory cache and must stay
* open for the next attempt -- and no append can be in flight, because acknowledgements are
* queued until the handle is ready.
*/
private void failReplay(Throwable t) {
stopReplay();
bufferedWriter.close();
pendingAckReplyCallBack.replayFailed(t);
}
}

private List<PendingAckMetadataEntry> deserializeEntry(Entry entry){
Expand All @@ -476,47 +544,160 @@ private List<PendingAckMetadataEntry> deserializeEntry(Entry entry){
class FillEntryQueueCallback implements AsyncCallbacks.ReadEntriesCallback {

private volatile boolean isReadable = true;
/**
* The read failure that ended this replay attempt, or null if no read failed. Written on a
* managed ledger thread by {@link #readEntriesFailed} before the volatile {@link #isReadable}
* write that stops the replay loop, and consumed by the replay thread only after it has
* observed {@code fillQueue() == false}, so the volatile hand-off publishes it. Never set by
* the failures that complete the replay instead (see {@link #readEntriesFailed}).
*/
private volatile ManagedLedgerException replayAttemptFailure;
private final AtomicLong outstandingReadsRequests = new AtomicLong(0);
private static final int NUMBER_OF_PER_READ_ENTRY = 100;
/**
* Guards {@link #stopped} against {@link #readEntriesComplete}. A read can still be in flight when
* the replay ends, and its completion runs on a managed ledger thread, so handing ownership of the
* entries over has to be atomic with respect to enqueuing them.
*/
private final Object stopLock = new Object();
private boolean stopped;

boolean fillQueue() {
// isReadable is deliberately checked AFTER outstandingReadsRequests: readEntriesFailed
// clears isReadable before its decrement, so a thread that observes the failed read's
// decrement here is guaranteed to also observe isReadable == false. No read is therefore
// ever issued after a read failure, which keeps the failure recorded by readEntriesFailed
// the only one of the attempt. (The reverse order would allow one more doomed read.)
if (entryQueue.size() + NUMBER_OF_PER_READ_ENTRY < entryQueue.capacity()
&& outstandingReadsRequests.get() == 0) {
&& outstandingReadsRequests.get() == 0
&& isReadable) {
if (cursor.hasMoreEntries()) {
outstandingReadsRequests.incrementAndGet();
readAsync(NUMBER_OF_PER_READ_ENTRY, this);
} else if (entryQueue.size() == 0) {
// Nothing left to read and everything read so far has been processed: the replay is
// done. The loop condition in PendingAckReplay cannot detect this on its own because
// it compares lastConfirmedEntry -- a snapshot taken when this store was created --
// against currentLoadPosition, which starts at the cursor's mark-delete position,
// while whether anything can still be read is decided by the cursor's read position.
// Those two can disagree permanently, e.g. when the ledger holding the mark-delete
// position was trimmed and the cursor was recovered onto a later ledger. Entries
// below the mark-delete position have already been applied, so completing here is
// correct. TopicTransactionBuffer's equivalent loop was fixed the same way in
// https://github.com/apache/pulsar/pull/13739
log.debug()
.attr("lastConfirmedEntry", lastConfirmedEntry)
.attr("currentLoadPosition", currentLoadPosition)
.attr("markDeletePosition", cursor.getMarkDeletedPosition())
.attr("readPosition", cursor.getReadPosition())
.log("Pending ack replay stopped before reaching the last confirmed entry "
+ "because the cursor has nothing left to read");
isReadable = false;
}
}
return isReadable;
}

@Override
public void readEntriesComplete(List<Entry> entries, Object ctx) {
entryQueue.fill(new MessagePassingQueue.Supplier<Entry>() {
private int i = 0;
@Override
public Entry get() {
Entry entry = entries.get(i);
i++;
return entry;
List<Entry> entriesToRelease = null;
synchronized (stopLock) {
if (stopped) {
// The replay already finished, so nothing will ever consume these.
entriesToRelease = entries;
} else {
int filled = entryQueue.fill(new MessagePassingQueue.Supplier<Entry>() {
private int i = 0;
@Override
public Entry get() {
Entry entry = entries.get(i);
i++;
return entry;
}
}, entries.size());
if (filled < entries.size()) {
entriesToRelease = entries.subList(filled, entries.size());
}
}
}, entries.size());

}
// Released outside the lock: releasing an entry can run a deallocation callback.
if (entriesToRelease != null) {
entriesToRelease.forEach(Entry::release);
}
outstandingReadsRequests.decrementAndGet();
}

/**
* Stops accepting entries and returns everything still queued, so that the replay thread can
* release them. Must only be called by the replay thread, which is the queue's single consumer.
*/
List<Entry> stopAndDrain() {
List<Entry> drained = new ArrayList<>();
synchronized (stopLock) {
stopped = true;
Entry entry;
while ((entry = entryQueue.poll()) != null) {
drained.add(entry);
}
}
return drained;
}

@Override
public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
if (managedLedger.getConfig().isAutoSkipNonRecoverableData()
// These three failures have always completed the replay as if it had reached the end of
// the log: the unreadable data is skipped by explicit configuration, or the store is being
// taken over or torn down and the handle must still reach Ready to release its callers.
// TransactionTest#testEndTPRecoveringWhenManagerLedgerDisReadable pins Ready for each.
boolean completesReplay = managedLedger.getConfig().isAutoSkipNonRecoverableData()
&& exception instanceof ManagedLedgerException.NonRecoverableLedgerException
|| exception instanceof ManagedLedgerException.ManagedLedgerFencedException
|| exception instanceof ManagedLedgerException.CursorAlreadyClosedException) {
isReadable = false;
|| exception instanceof ManagedLedgerException.CursorAlreadyClosedException;
if (!completesReplay) {
// Any other failure ends this replay attempt as failed instead of re-issuing the same
// read in a hot loop that monopolises the shared replay thread (issue #26374).
// PendingAckHandleImpl#exceptionHandleFuture classifies what happens next: transient
// failures (a plain ManagedLedgerException, e.g. a BookKeeper outage, a read timeout
// or read throttling) reschedule init() with backoff, which frees the replay thread
// between attempts; permanent ones (NonRecoverableLedgerException with
// autoSkipNonRecoverableData disabled) fail the subscription fast instead of retrying
// a read that can never succeed. The volatile write below must stay before the
// isReadable write: observing isReadable == false is what publishes it to the replay
// thread.
replayAttemptFailure = exception;
}
log.error().exception(exception).log("MLPendingAckStore of topic stat reply fail");
if (completesReplay) {
// This path never reaches exceptionHandleFuture, and the handle goes on to log its
// recovery as a success, so without this line a store fenced by a takeover, a cursor
// closed by teardown, or data skipped because autoSkipNonRecoverableData is enabled
// would leave no record at any default level -- a cursor closed between two reads
// fails synchronously in ManagedCursorImpl without even a managed ledger layer log.
// At most one line per replay attempt: only one read is ever outstanding.
log.warn()
.exception(exception)
.attr("currentLoadPosition", currentLoadPosition)
.attr("lastConfirmedEntry", lastConfirmedEntry)
.log("Pending ack replay read failed; completing the replay anyway");
} else {
// The record operators act on is written downstream by
// PendingAckHandleImpl#exceptionHandleFuture: one WARN per backoff paced retry, or an
// ERROR when it gives up and fails the subscription.
log.debug().exception(exception).log("Pending ack replay read failed");
}
// Written before the decrement so that fillQueue() can never issue another read after a
// failure: it re-checks isReadable after observing outstandingReadsRequests == 0.
isReadable = false;
outstandingReadsRequests.decrementAndGet();
}

/**
* The read failure that ended this replay attempt, or null if it ended without one. Only
* meaningful once the replay loop has exited.
*/
ManagedLedgerException replayAttemptFailure() {
return replayAttemptFailure;
}

}

public CompletableFuture<ManagedLedger> getManagedLedger() {
Expand Down
5 changes: 4 additions & 1 deletion .../main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -983,7 +983,10 @@ public void completeHandleFuture() {
}

public void exceptionHandleFuture(Throwable t) {
if (isRetryableException(t)) {
// Never retry once the handle is closing or closed. The retry path below resets the state to
// None before scheduling init(), which would defeat the checkIfClose() guard in
// initPendingAckStore() and reopen the pending ack store of a subscription that is going away.
if (isRetryableException(t) && !checkIfClose()) {
this.state = State.None;
long retryTime = backoff.next().toMillis();
log.warn()
Expand Down
Loading
Loading