diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java index c3524e1c2ea0f..e005873b83983 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java @@ -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; @@ -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 deserializeEntry(Entry entry){ @@ -476,15 +544,55 @@ private List 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; @@ -492,31 +600,104 @@ boolean fillQueue() { @Override public void readEntriesComplete(List entries, Object ctx) { - entryQueue.fill(new MessagePassingQueue.Supplier() { - private int i = 0; - @Override - public Entry get() { - Entry entry = entries.get(i); - i++; - return entry; + List 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() { + 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 stopAndDrain() { + List 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 getManagedLedger() { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java index c417fd6d18e75..9649f04e71e58 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java @@ -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() diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java index b20e44d6dc09a..09840a42c9ea6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTest.java @@ -18,22 +18,41 @@ */ package org.apache.pulsar.broker.transaction.pendingack.impl; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.same; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import io.netty.util.Timeout; +import io.netty.util.Timer; import java.util.ArrayList; import java.util.Iterator; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import lombok.CustomLog; +import org.apache.bookkeeper.mledger.AsyncCallbacks; +import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.ManagedLedger; +import org.apache.bookkeeper.mledger.ManagedLedgerConfig; +import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.pulsar.broker.ServiceConfiguration; @@ -49,6 +68,7 @@ import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.util.Codec; import org.apache.pulsar.common.util.FutureUtil; +import org.apache.pulsar.transaction.coordinator.impl.DisabledTxnLogBufferedWriterMetricsStats; import org.apache.pulsar.transaction.coordinator.impl.TxnLogBufferedWriterConfig; import org.awaitility.Awaitility; import org.mockito.invocation.InvocationOnMock; @@ -170,6 +190,365 @@ public void testPendingAckStoreWithSlashSubscriptionName() throws Exception { closePendingAckStoreWithRetry(store); } + private MLPendingAckStore createPendingAckStoreForReplay(ManagedCursor cursor) { + return createPendingAckStoreForReplay(cursor, new ManagedLedgerConfig()); + } + + /** + * Builds a store whose replay loop starts in the state described by the parameters, without going + * through the provider: the replay guard compares {@code lastConfirmedEntry} (a snapshot of the + * managed ledger's last confirmed entry, taken here) against a load position seeded from the + * cursor's mark-delete position, while whether anything can still be read is decided separately by + * {@link ManagedCursor#hasMoreEntries()}. + */ + private MLPendingAckStore createPendingAckStoreForReplay(ManagedCursor cursor, ManagedLedgerConfig mlConfig) { + return createPendingAckStoreForReplay(cursor, mlConfig, mock(Timer.class), false); + } + + private MLPendingAckStore createPendingAckStoreForReplay(ManagedCursor cursor, ManagedLedgerConfig mlConfig, + Timer timer, boolean batchEnabled) { + ManagedLedger managedLedger = mock(ManagedLedger.class); + when(managedLedger.getName()).thenReturn("test-pending-ack-log"); + when(managedLedger.getConfig()).thenReturn(mlConfig); + // Deliberately ahead of the cursor's mark-delete position, so the replay loop's guard stays true. + when(managedLedger.getLastConfirmedEntry()).thenReturn(PositionFactory.create(5, 10)); + TxnLogBufferedWriterConfig config = new TxnLogBufferedWriterConfig(); + config.setBatchEnabled(batchEnabled); + return new MLPendingAckStore(managedLedger, cursor, mock(ManagedCursor.class), 1, config, + timer, DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS, + internalPinnedExecutor); + } + + private ManagedCursor createReplayCursorMock() { + ManagedCursor cursor = mock(ManagedCursor.class); + when(cursor.getName()).thenReturn("test-pending-ack-cursor"); + when(cursor.getMarkDeletedPosition()).thenReturn(PositionFactory.create(1, 0)); + return cursor; + } + + /** + * The replay loop must finish once the cursor has nothing left to read, even though its termination + * guard still believes there is work to do. Those two are measured from different positions -- the + * guard from the cursor's mark-delete position, the read gate from its read position -- so they can + * disagree permanently, for instance after the ledger holding the mark-delete position was trimmed + * and the cursor was recovered onto a later ledger. Before the fix the loop spun on Thread.sleep(1) + * forever, holding its executor thread and starving every other subscription hashed onto it. + */ + @Test + public void testReplayCompletesWhenCursorHasNoMoreEntries() throws Exception { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.isClosed()).thenReturn(false); + // No entry between the mark-delete position and the snapshot can be read any more. + when(cursor.hasMoreEntries()).thenReturn(false); + MLPendingAckStore pendingAckStore = createPendingAckStoreForReplay(cursor); + + ExecutorService replayExecutor = Executors.newSingleThreadExecutor(); + try { + PendingAckHandleImpl pendingAckHandle = mock(PendingAckHandleImpl.class); + when(pendingAckHandle.getInternalPinnedExecutor()).thenReturn(replayExecutor); + when(pendingAckHandle.changeToReadyState()).thenReturn(true); + CountDownLatch replayFinished = new CountDownLatch(1); + doAnswer(invocation -> { + replayFinished.countDown(); + return null; + }).when(pendingAckHandle).completeHandleFuture(); + + pendingAckStore.replayAsync(pendingAckHandle, replayExecutor); + + // The replay thread is shared by every subscription hashed onto it, so a task queued behind a + // finished replay must still get to run. This is what the stall actually broke. + CountDownLatch queuedBehindReplay = new CountDownLatch(1); + replayExecutor.execute(queuedBehindReplay::countDown); + + Assert.assertTrue(replayFinished.await(10, TimeUnit.SECONDS), + "Replay never completed: the replay loop did not terminate"); + Assert.assertTrue(queuedBehindReplay.await(10, TimeUnit.SECONDS), + "A task queued behind the replay never ran: the replay thread was not released"); + } finally { + replayExecutor.shutdownNow(); + } + } + + /** + * A read whose completion never arrives leaves the replay loop waiting with no way to make progress. + * Closing the cursor -- which is what unloading the topic does -- must stop it, so that the replay + * thread cannot outlive the subscription it belongs to. + */ + @Test + public void testReplayStopsWhenCursorIsClosedWhileWaitingForEntries() throws Exception { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.hasMoreEntries()).thenReturn(true); + AtomicBoolean cursorClosed = new AtomicBoolean(false); + when(cursor.isClosed()).thenAnswer(invocation -> cursorClosed.get()); + CountDownLatch readIssued = new CountDownLatch(1); + // Drop the read: neither readEntriesComplete nor readEntriesFailed is ever invoked, so the loop + // keeps waiting for entries that never arrive. + doAnswer(invocation -> { + readIssued.countDown(); + return null; + }).when(cursor).asyncReadEntries(anyInt(), any(), any(), any()); + MLPendingAckStore pendingAckStore = createPendingAckStoreForReplay(cursor); + + ExecutorService replayExecutor = Executors.newSingleThreadExecutor(); + try { + PendingAckHandleImpl pendingAckHandle = mock(PendingAckHandleImpl.class); + when(pendingAckHandle.getInternalPinnedExecutor()).thenReturn(replayExecutor); + + pendingAckStore.replayAsync(pendingAckHandle, replayExecutor); + Assert.assertTrue(readIssued.await(10, TimeUnit.SECONDS), "Replay never issued a read"); + + cursorClosed.set(true); + + verify(pendingAckHandle, timeout(TimeUnit.SECONDS.toMillis(10))) + .exceptionHandleFuture(any(ManagedLedgerException.CursorAlreadyClosedException.class)); + // The replay must have released the thread it was holding. + CountDownLatch queuedBehindReplay = new CountDownLatch(1); + replayExecutor.execute(queuedBehindReplay::countDown); + Assert.assertTrue(queuedBehindReplay.await(10, TimeUnit.SECONDS), + "A task queued behind the replay never ran: the replay thread was not released"); + } finally { + replayExecutor.shutdownNow(); + } + } + + /** + * A read can still be in flight when the replay ends, and its completion runs on a managed ledger + * thread. Entries delivered after the replay has gone must be released by the callback itself, + * because nothing will ever take them off the queue. + */ + @Test + public void testEntriesDeliveredAfterReplayEndedAreReleased() throws Exception { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.hasMoreEntries()).thenReturn(true); + AtomicBoolean cursorClosed = new AtomicBoolean(false); + when(cursor.isClosed()).thenAnswer(invocation -> cursorClosed.get()); + CountDownLatch readIssued = new CountDownLatch(1); + AtomicReference readCallback = new AtomicReference<>(); + // Capture the callback and never complete it, so the read is still in flight when the replay ends. + doAnswer(invocation -> { + readCallback.set(invocation.getArgument(1)); + readIssued.countDown(); + return null; + }).when(cursor).asyncReadEntries(anyInt(), any(), any(), any()); + MLPendingAckStore pendingAckStore = createPendingAckStoreForReplay(cursor); + + ExecutorService replayExecutor = Executors.newSingleThreadExecutor(); + try { + PendingAckHandleImpl pendingAckHandle = mock(PendingAckHandleImpl.class); + when(pendingAckHandle.getInternalPinnedExecutor()).thenReturn(internalPinnedExecutor); + + pendingAckStore.replayAsync(pendingAckHandle, replayExecutor); + Assert.assertTrue(readIssued.await(10, TimeUnit.SECONDS), "Replay never issued a read"); + + // End the replay while the read is still outstanding. + cursorClosed.set(true); + verify(pendingAckHandle, timeout(TimeUnit.SECONDS.toMillis(10))) + .exceptionHandleFuture(any(ManagedLedgerException.CursorAlreadyClosedException.class)); + + // The read now completes, far too late for the replay to consume anything. + Entry first = mock(Entry.class); + Entry second = mock(Entry.class); + readCallback.get().readEntriesComplete(List.of(first, second), null); + + verify(first).release(); + verify(second).release(); + } finally { + replayExecutor.shutdownNow(); + } + } + + /** + * Shutting down the replay executor must end the replay rather than leaving the thread behind. The + * replay is incomplete at that point, so it must be reported as failed and not as complete -- + * otherwise the handle would be marked Ready with only part of the pending ack state applied. + */ + @Test + public void testReplayStopsWhenInterrupted() throws Exception { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.isClosed()).thenReturn(false); + when(cursor.hasMoreEntries()).thenReturn(true); + CountDownLatch readIssued = new CountDownLatch(1); + // Drop the read, so the replay is waiting for entries that never arrive when it is interrupted. + doAnswer(invocation -> { + readIssued.countDown(); + return null; + }).when(cursor).asyncReadEntries(anyInt(), any(), any(), any()); + MLPendingAckStore pendingAckStore = createPendingAckStoreForReplay(cursor); + + ExecutorService replayExecutor = Executors.newSingleThreadExecutor(); + try { + PendingAckHandleImpl pendingAckHandle = mock(PendingAckHandleImpl.class); + when(pendingAckHandle.getInternalPinnedExecutor()).thenReturn(internalPinnedExecutor); + + pendingAckStore.replayAsync(pendingAckHandle, replayExecutor); + Assert.assertTrue(readIssued.await(10, TimeUnit.SECONDS), "Replay never issued a read"); + + replayExecutor.shutdownNow(); + + Assert.assertTrue(replayExecutor.awaitTermination(10, TimeUnit.SECONDS), + "The replay thread did not stop after shutdownNow()"); + verify(pendingAckHandle, timeout(TimeUnit.SECONDS.toMillis(10))) + .exceptionHandleFuture(any(InterruptedException.class)); + // An interrupted replay is incomplete and must never be reported as successful. + verify(pendingAckHandle, never()).completeHandleFuture(); + } finally { + replayExecutor.shutdownNow(); + } + } + + @DataProvider(name = "replayEndingReadFailures") + public Object[][] replayEndingReadFailuresProvider() { + return new Object[][]{ + // A deleted pending ack ledger with autoSkipNonRecoverableData disabled (the default): + // retrying can never succeed, so the subscription must fail fast. + {new ManagedLedgerException.LedgerNotExistException("Ledger does not exist")}, + // A transient BookKeeper failure ("Bookie handle is not available", BK code -8). The + // configured read timeout surfaces as this same plain class. The retry belongs in + // PendingAckHandleImpl's backoff-paced init(), not in a hot read loop. + {new ManagedLedgerException("Bookie handle is not available")}, + // Read throttling: also transient, also paced by the handle's backoff. + {new ManagedLedgerException.TooManyRequestsException("Too many concurrent reads")}, + }; + } + + /** + * A read failure outside the historically recognised set must end the replay attempt through + * {@code replayFailed}, handing the original exception unwrapped to + * {@link PendingAckHandleImpl#exceptionHandleFuture}, which either retries with backoff (transient + * failures) or fails the subscription fast (non-recoverable data without auto-skip). Before the fix + * these exceptions did not stop the loop: the identical read was re-issued forever, monopolising + * the replay thread shared by every subscription hashed onto it (issue #26374). + */ + @Test(dataProvider = "replayEndingReadFailures") + public void testReadFailureEndsReplayAttempt(ManagedLedgerException failure) throws Exception { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.isClosed()).thenReturn(false); + when(cursor.hasMoreEntries()).thenReturn(true); + doAnswer(invocation -> { + AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(1); + callback.readEntriesFailed(failure, null); + return null; + }).when(cursor).asyncReadEntries(anyInt(), any(), any(), any()); + MLPendingAckStore pendingAckStore = createPendingAckStoreForReplay(cursor); + + ExecutorService replayExecutor = Executors.newSingleThreadExecutor(); + try { + PendingAckHandleImpl pendingAckHandle = mock(PendingAckHandleImpl.class); + when(pendingAckHandle.getInternalPinnedExecutor()).thenReturn(replayExecutor); + + pendingAckStore.replayAsync(pendingAckHandle, replayExecutor); + + // The original exception must arrive unwrapped: isRetryableException discriminates on the + // concrete class. + verify(pendingAckHandle, timeout(TimeUnit.SECONDS.toMillis(10))).exceptionHandleFuture(same(failure)); + // The replay thread is shared: a task queued behind the ended replay must get to run. + CountDownLatch queuedBehindReplay = new CountDownLatch(1); + replayExecutor.execute(queuedBehindReplay::countDown); + Assert.assertTrue(queuedBehindReplay.await(10, TimeUnit.SECONDS), + "A task queued behind the replay never ran: the replay thread was not released"); + // A failed attempt must never also be reported successful. Sequenced after the queued task, + // so a wrongly scheduled replayComplete would already have run on replayExecutor. + verify(pendingAckHandle, never()).completeHandleFuture(); + // One read, one failure, one outcome: the store must not retry the read itself. + verify(cursor, times(1)).asyncReadEntries(anyInt(), any(), any(), any()); + // The cursor is cached and reused by the retry, so a failed attempt must rewind it: reads + // run ahead of processing, and entries this attempt read but never applied were released. + verify(cursor).rewind(); + } finally { + replayExecutor.shutdownNow(); + } + } + + @DataProvider(name = "replayCompletingReadFailures") + public Object[][] replayCompletingReadFailuresProvider() { + return new Object[][]{ + {new ManagedLedgerException.NonRecoverableLedgerException("No ledger exist"), true}, + {new ManagedLedgerException.ManagedLedgerFencedException(), false}, + {new ManagedLedgerException.CursorAlreadyClosedException("Cursor already closed"), false}, + }; + } + + /** + * The three historically recognised read failures keep completing the replay: non-recoverable data + * is skipped when autoSkipNonRecoverableData is the operator's explicit choice, and a fenced + * managed ledger or an already-closed cursor means the store is being taken over or torn down, + * where the handle must still reach Ready to release its callers. Store-level pin of the behaviour + * TransactionTest#testEndTPRecoveringWhenManagerLedgerDisReadable asserts end to end; a fix that + * routed every read failure to replayFailed would fail this. + */ + @Test(dataProvider = "replayCompletingReadFailures") + public void testReadFailuresThatCompleteTheReplay(ManagedLedgerException failure, boolean autoSkip) + throws Exception { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.isClosed()).thenReturn(false); + when(cursor.hasMoreEntries()).thenReturn(true); + doAnswer(invocation -> { + AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(1); + callback.readEntriesFailed(failure, null); + return null; + }).when(cursor).asyncReadEntries(anyInt(), any(), any(), any()); + ManagedLedgerConfig mlConfig = new ManagedLedgerConfig(); + mlConfig.setAutoSkipNonRecoverableData(autoSkip); + MLPendingAckStore pendingAckStore = createPendingAckStoreForReplay(cursor, mlConfig); + + ExecutorService replayExecutor = Executors.newSingleThreadExecutor(); + try { + PendingAckHandleImpl pendingAckHandle = mock(PendingAckHandleImpl.class); + when(pendingAckHandle.getInternalPinnedExecutor()).thenReturn(replayExecutor); + when(pendingAckHandle.changeToReadyState()).thenReturn(true); + CountDownLatch replayFinished = new CountDownLatch(1); + doAnswer(invocation -> { + replayFinished.countDown(); + return null; + }).when(pendingAckHandle).completeHandleFuture(); + + pendingAckStore.replayAsync(pendingAckHandle, replayExecutor); + + Assert.assertTrue(replayFinished.await(10, TimeUnit.SECONDS), + "Replay did not complete for " + failure.getClass().getSimpleName()); + verify(pendingAckHandle, never()).exceptionHandleFuture(any()); + // A replay that completes must leave the shared cursor's read position alone. + verify(cursor, never()).rewind(); + } finally { + replayExecutor.shutdownNow(); + } + } + + /** + * A failed replay attempt orphans its store, because the retry builds a new one. It must therefore + * close its buffered writer, whose timing flush task otherwise reschedules itself forever when + * pending ack batching is enabled and nothing is left holding a reference that could stop it. + */ + @Test + public void testFailedReplayAttemptClosesBufferedWriter() throws Exception { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.isClosed()).thenReturn(false); + when(cursor.hasMoreEntries()).thenReturn(true); + doAnswer(invocation -> { + AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(1); + callback.readEntriesFailed(new ManagedLedgerException("Bookie handle is not available"), null); + return null; + }).when(cursor).asyncReadEntries(anyInt(), any(), any(), any()); + Timer timer = mock(Timer.class); + Timeout flushTimeout = mock(Timeout.class); + when(timer.newTimeout(any(), anyLong(), any())).thenReturn(flushTimeout); + MLPendingAckStore pendingAckStore = + createPendingAckStoreForReplay(cursor, new ManagedLedgerConfig(), timer, true); + + ExecutorService replayExecutor = Executors.newSingleThreadExecutor(); + try { + PendingAckHandleImpl pendingAckHandle = mock(PendingAckHandleImpl.class); + when(pendingAckHandle.getInternalPinnedExecutor()).thenReturn(replayExecutor); + + pendingAckStore.replayAsync(pendingAckHandle, replayExecutor); + + verify(pendingAckHandle, timeout(TimeUnit.SECONDS.toMillis(10))).exceptionHandleFuture(any()); + verify(flushTimeout, timeout(TimeUnit.SECONDS.toMillis(10))).cancel(); + } finally { + replayExecutor.shutdownNow(); + } + } + /** * Overridden cases: * 1. Batched write and replay with batched feature.