From 5da5f22f501afe8817fd6d1c2bc56d9be262b7b9 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Tue, 18 Aug 2026 13:37:22 +0300 Subject: [PATCH 1/2] [fix][txn] Stop the pending ack replay loop from spinning forever Fixes #26368 ### Motivation `MLPendingAckStore.PendingAckReplay.run()` loops while `lastConfirmedEntry.compareTo(currentLoadPosition) > 0 && fillEntryQueueCallback.fillQueue()`, sleeping 1ms whenever the entry queue is empty. The two halves of that condition are measured from different positions: `lastConfirmedEntry` is a snapshot of the managed ledger's last confirmed entry taken in the constructor and is compared against `currentLoadPosition`, which starts at the cursor's mark-delete position, while whether anything can still be read is decided by `cursor.hasMoreEntries()`, which follows the cursor's read position. When those disagree permanently, `fillQueue()` issues no read but still returns `isReadable == true`, so the loop sleeps forever with no outstanding read and nothing logged. One way to reach it: the cursor persists a mark-delete position at the last entry of a ledger, that ledger is trimmed, and after a restart `ManagedCursorImpl.recoveredCursor` leaves the stored position unrepaired (it only substitutes when `entryId == -1`) while the read position moves to a later ledger. The replay executors are single threaded and assigned by hash, so a stuck replay also stops every other subscription sharing that thread from adding consumers: with transactions enabled every persistent subscription builds a `PendingAckHandle`, and `PersistentSubscription.addConsumerInternal` waits on `pendingAckHandleFuture()` with no timeout. The loop could not be stopped either. `cursor.isClosed()` was only checked before the loop, so a read whose completion never arrives could not be ended by closing the subscription, and the `InterruptedException` handler only logged, so the thread survived `ExecutorProvider.shutdownNow()`. `TopicTransactionBuffer` has the same recovery loop and was fixed for this in #13739 (commit 7dee63ed707), but the fix was never ported to `MLPendingAckStore`. ### Modifications - `FillEntryQueueCallback.fillQueue()`: when the cursor has no more entries and the queue is drained, set `isReadable = false` so the replay finishes. This mirrors the existing `TopicTransactionBuffer` implementation. Entries at or below the mark-delete position have already been applied, so completing here is correct. - `PendingAckReplay.run()`: re-check `cursor.isClosed()` while waiting for entries, so that closing the subscription ends a replay whose read completion never arrives. A cursor closed between two reads already ends the loop through the existing read failure handling, because that read fails synchronously. - `PendingAckReplay.run()`: on `InterruptedException`, restore the interrupt flag and end the replay through `replayFailed()` rather than continuing. An incomplete replay must not be reported as successful, and the task is now cancellable. - `PendingAckReplay.run()`: end every exit through a single `stopReplay()` step that marks the callback stopped and releases entries still queued. A read issued by `fillQueue()` can still be outstanding when the replay ends, and its completion runs on a managed ledger thread, so `readEntriesComplete` now releases entries directly instead of queueing them once the replay has stopped. The handover is done under a lock so the check and the enqueue cannot interleave, and entries are released outside it because releasing can run a deallocation callback. The queue keeps a single consumer: late callbacks release their own entries and never poll. This covers the normal `replayComplete()` exit too, which could already leak entries read past the point the replay needed. - `PendingAckHandleImpl.exceptionHandleFuture()`: do not retry once the handle is closing or closed. The retry path resets the state to `None` before scheduling `init()`, which defeats the `checkIfClose()` guard in `initPendingAckStore()` and reopens the pending ack store of a subscription that is going away. This was reachable before, and the new in-loop close check makes it reachable on the ordinary topic unload path. ### Verifying this change Three tests are added to `MLPendingAckStoreTest`. Each was verified to fail when the corresponding change is reverted. - `testReplayCompletesWhenCursorHasNoMoreEntries` builds the diverged state and asserts both that the replay completes and that a task queued behind it on the same executor still runs. - `testReplayStopsWhenCursorIsClosedWhileWaitingForEntries` drops a read completion, closes the cursor, and asserts the replay fails and releases the thread. - `testReplayStopsWhenInterrupted` drops a read completion, calls `shutdownNow()`, and asserts the thread terminates, that `replayFailed` is invoked, and that the replay is never reported as complete. - `testEntriesDeliveredAfterReplayEndedAreReleased` ends the replay with a read still in flight, then completes that read and asserts every delivered entry was released. ### Documentation - [x] `doc-not-needed` Assisted-by: Claude (Opus 5, Fable), OpenAI Codex (gpt-5.6-sol) --- .../pendingack/impl/MLPendingAckStore.java | 115 ++++++++-- .../pendingack/impl/PendingAckHandleImpl.java | 5 +- .../impl/MLPendingAckStoreTest.java | 211 ++++++++++++++++++ 3 files changed, 316 insertions(+), 15 deletions(-) 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..b005ca5a5f264 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 @@ -434,24 +434,55 @@ 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()) { + stopReplay(); + pendingAckReplyCallBack.replayFailed(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(); + stopReplay(); + pendingAckReplyCallBack.replayFailed(e); + log.warn() + .exception(e) + .log("Transaction pending ack replay thread was interrupted"); + return; } } } } catch (Exception e) { + stopReplay(); pendingAckReplyCallBack.replayFailed(e); log.error().exception(e).log("Pending ack recover fail"); return; } + stopReplay(); 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(); + } + } } private List deserializeEntry(Entry entry){ @@ -478,6 +509,13 @@ class FillEntryQueueCallback implements AsyncCallbacks.ReadEntriesCallback { private volatile boolean isReadable = true; 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() { if (entryQueue.size() + NUMBER_OF_PER_READ_ENTRY < entryQueue.capacity() @@ -485,6 +523,25 @@ boolean fillQueue() { 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,19 +549,49 @@ 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() 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..ddbf2de72ab75 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,36 @@ */ package org.apache.pulsar.broker.transaction.pendingack.impl; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; 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.verify; import static org.mockito.Mockito.when; +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.ManagedLedgerException; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.pulsar.broker.ServiceConfiguration; @@ -49,6 +63,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 +185,202 @@ public void testPendingAckStoreWithSlashSubscriptionName() throws Exception { closePendingAckStoreWithRetry(store); } + /** + * 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) { + ManagedLedger managedLedger = mock(ManagedLedger.class); + when(managedLedger.getName()).thenReturn("test-pending-ack-log"); + // 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(false); + return new MLPendingAckStore(managedLedger, cursor, mock(ManagedCursor.class), 1, config, + mock(Timer.class), 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(); + } + } + /** * Overridden cases: * 1. Batched write and replay with batched feature. From 52f030b2e7c4f8d14806dfe6cfd42d4811d1d09c Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 19 Aug 2026 03:32:26 +0300 Subject: [PATCH 2/2] [fix][txn] End pending ack replay on read failures it does not classify Fixes #26374 ### Motivation `MLPendingAckStore.FillEntryQueueCallback.readEntriesFailed` recognises three exception shapes and ends the replay for them. Everything else only logged an ERROR and decremented the outstanding read counter, so `fillQueue()` immediately re-issued the identical read. Two classes fall through, and both are reachable with the shipped defaults: - `NonRecoverableLedgerException` when `autoSkipNonRecoverableData` is false. A deleted or missing pending ack ledger arrives as `LedgerNotExistException`, and `OpReadEntry` only advances past a bad ledger when auto-skip is enabled, so the read position never moves and every retry is identical. Retrying can never succeed. - A plain `ManagedLedgerException`, such as BookKeeper's `BookieHandleNotAvailableException` (code -8). Retrying is right here, but not at this rate. The result is a hot loop of roughly a thousand reads per second, one ERROR line each, holding a transaction replay thread. Those executors are single threaded and assigned by hash, and `PersistentSubscription.addConsumerInternal` waits on `pendingAckHandleFuture()` with no timeout, so every other subscription on that thread stops being able to add consumers. The reporter of #26364 measured roughly 47.6k, 1.5k, 59.4k and 31.1k of these ERRORs across four brokers in fifteen minutes during a BookKeeper outage, and reproduced the deleted ledger case deterministically: an unrelated topic sharing the replay thread also stopped accepting subscriptions. This is the half of #12700 that was never finished. That PR set out to stop exactly this ("if any ledger was deleted from bookkeeper ... MLPendingAckStore will not stop recovering and continue to report the exception") and added both the guard and `TransactionTest.testEndTPRecoveringWhenManagerLedgerDisReadable`. But that test enables `autoSkipNonRecoverableData`, so the default configuration was left spinning. ### Modifications - `readEntriesFailed` now ends the replay for every failure. The three shapes it already recognised keep completing the replay, exactly as the test above pins; any other failure is recorded and the attempt is reported through `replayFailed`, letting `PendingAckHandleImpl.exceptionHandleFuture` decide what happens next. It already classifies correctly: transient failures reschedule `init()` with backoff, which frees the replay thread between attempts, and non-recoverable ones fail the subscription instead of retrying a read that cannot succeed. - `fillQueue()` now checks `isReadable` after `outstandingReadsRequests`. `readEntriesFailed` clears `isReadable` before its decrement, so a thread that observes the decrement must also observe the cleared flag, and no read can be issued after a failure. The reverse operand order would allow one more. - A failed attempt rewinds the cursor. Reads run ahead of processing, so entries that were read but never applied have already advanced the shared cursor's read position and are released when the attempt ends. The cursor is cached by the managed ledger, so without a rewind the next attempt would resume past those entries, skip them, and report the incomplete replay as successful. - A failed attempt closes its buffered writer. The store is abandoned afterwards, and with `transactionPendingAckBatchedWriteEnabled` the writer's flush task reschedules itself forever, with nothing left able to stop it. - The per-failure log is split. Failures that complete the replay log at WARN, because they never reach `exceptionHandleFuture` and the handle goes on to log its recovery as a success; a cursor closed between two reads fails synchronously with no log at any layer, so this was otherwise invisible. Failures reported downstream drop to DEBUG, where `exceptionHandleFuture` writes one WARN per paced attempt or an ERROR when it gives up. Note for operators: the old `"MLPendingAckStore of topic ... stat reply fail!"` line is gone. The equivalents are the WARN above and the messages from `exceptionHandleFuture`. ### Verifying this change Added to `MLPendingAckStoreTest`, each verified to fail when its change is reverted: - `testReadFailureEndsReplayAttempt`, over `LedgerNotExistException`, a plain `ManagedLedgerException` and `TooManyRequestsException`: the original exception reaches `exceptionHandleFuture` unwrapped, exactly one read is issued, the replay thread is released, the replay is never also reported complete, and the cursor is rewound. - `testReadFailuresThatCompleteTheReplay`, over the three recognised shapes: each still completes, none reaches `exceptionHandleFuture`, and the cursor is not rewound. This pins the discriminating shape; routing every failure to `replayFailed` fails it. - `testFailedReplayAttemptClosesBufferedWriter`: the flush timer is cancelled. `MLPendingAckStoreTest` (16) and `TransactionTest` (36) pass, including `testEndTPRecoveringWhenManagerLedgerDisReadable`. ### Documentation - [x] `doc-not-needed` Assisted-by: Claude (Opus 5, Fable), OpenAI Codex (gpt-5.6-sol) --- .../pendingack/impl/MLPendingAckStore.java | 118 ++++++++++-- .../impl/MLPendingAckStoreTest.java | 174 +++++++++++++++++- 2 files changed, 277 insertions(+), 15 deletions(-) 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 b005ca5a5f264..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; @@ -440,8 +440,7 @@ public void run() { // ManagedCursorImpl#asyncReadEntriesWithSkip, which clears isReadable and ends // the loop through the existing read failure handling. if (cursor.isClosed()) { - stopReplay(); - pendingAckReplyCallBack.replayFailed(new ManagedLedgerException + failReplay(new ManagedLedgerException .CursorAlreadyClosedException("MLPendingAckStore cursor was closed " + "while replaying.")); log.warn("MLPendingAckStore cursor was closed while replaying, close replay thread"); @@ -454,8 +453,7 @@ public void run() { // be reported as successful: replayFailed() lets PendingAckHandleImpl decide // whether to retry instead of leaving the handle Ready with partial state. Thread.currentThread().interrupt(); - stopReplay(); - pendingAckReplyCallBack.replayFailed(e); + failReplay(e); log.warn() .exception(e) .log("Transaction pending ack replay thread was interrupted"); @@ -464,12 +462,36 @@ public void run() { } } } catch (Exception e) { - stopReplay(); - 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(); } @@ -483,6 +505,21 @@ private void stopReplay() { 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){ @@ -507,6 +544,14 @@ 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; /** @@ -518,8 +563,14 @@ class FillEntryQueueCallback implements AsyncCallbacks.ReadEntriesCallback { 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); @@ -594,16 +645,59 @@ List stopAndDrain() { @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/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 ddbf2de72ab75..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 @@ -20,13 +20,17 @@ 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; @@ -47,6 +51,7 @@ 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; @@ -185,6 +190,10 @@ 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 @@ -192,15 +201,21 @@ public void testPendingAckStoreWithSlashSubscriptionName() throws Exception { * cursor's mark-delete position, while whether anything can still be read is decided separately by * {@link ManagedCursor#hasMoreEntries()}. */ - private MLPendingAckStore createPendingAckStoreForReplay(ManagedCursor cursor) { + 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(false); + config.setBatchEnabled(batchEnabled); return new MLPendingAckStore(managedLedger, cursor, mock(ManagedCursor.class), 1, config, - mock(Timer.class), DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS, + timer, DisabledTxnLogBufferedWriterMetricsStats.DISABLED_BUFFERED_WRITER_METRICS, internalPinnedExecutor); } @@ -381,6 +396,159 @@ public void testReplayStopsWhenInterrupted() throws Exception { } } + @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.