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..c0c056e4509ce 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 @@ -413,45 +413,81 @@ public void run() { while (lastConfirmedEntry.compareTo(currentLoadPosition) > 0 && fillEntryQueueCallback.fillQueue()) { Entry entry = entryQueue.poll(); if (entry != null) { - currentLoadPosition = PositionFactory.create(entry.getLedgerId(), entry.getEntryId()); - List logs = deserializeEntry(entry); - if (logs.isEmpty()){ - continue; - } else if (logs.size() == 1){ - currentIndexLag.incrementAndGet(); - PendingAckMetadataEntry log = logs.get(0); - handleMetadataEntry(PositionFactory.create(entry.getLedgerId(), entry.getEntryId()), log); - pendingAckReplyCallBack.handleMetadataEntry(log); - } else { - int batchSize = logs.size(); - for (int batchIndex = 0; batchIndex < batchSize; batchIndex++){ - PendingAckMetadataEntry log = logs.get(batchIndex); + try { + currentLoadPosition = PositionFactory.create(entry.getLedgerId(), entry.getEntryId()); + List logs = deserializeEntry(entry); + if (logs.isEmpty()){ + continue; + } else if (logs.size() == 1){ + currentIndexLag.incrementAndGet(); + PendingAckMetadataEntry log = logs.get(0); + handleMetadataEntry(PositionFactory.create(entry.getLedgerId(), entry.getEntryId()), + log); pendingAckReplyCallBack.handleMetadataEntry(log); + } else { + int batchSize = logs.size(); + for (int batchIndex = 0; batchIndex < batchSize; batchIndex++){ + PendingAckMetadataEntry log = logs.get(batchIndex); + pendingAckReplyCallBack.handleMetadataEntry(log); + } + currentIndexLag.addAndGet(batchSize); + handleMetadataEntry(PositionFactory.create(entry.getLedgerId(), entry.getEntryId()), + logs); } - currentIndexLag.addAndGet(batchSize); - handleMetadataEntry(PositionFactory.create(entry.getLedgerId(), entry.getEntryId()), logs); + } finally { + entry.release(); } - 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 +514,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 +528,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 +554,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..3706cfad7110f 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 @@ -193,20 +193,29 @@ private void init() { } private void initPendingAckStore() { - if (changeToInitializingState()) { + CompletableFuture storeFuture; + synchronized (this) { + if (!changeToInitializingState()) { + return; + } + // Publish ownership before calling the provider so closeAsync() also waits for a store + // whose creation has started but whose future has not yet been returned. + storeFuture = new CompletableFuture<>(); + this.pendingAckStoreFuture = storeFuture; + } + storeFuture.thenAccept(pendingAckStore -> { if (!checkIfClose()) { - this.pendingAckStoreFuture = - pendingAckStoreProvider.newPendingAckStore(persistentSubscription); - this.pendingAckStoreFuture.thenAccept(pendingAckStore -> { - recoverTime.setRecoverStartTime(System.currentTimeMillis()); - pendingAckStore.replayAsync(this, internalPinnedExecutor); - }).exceptionallyAsync(e -> { - // Handling the exceptions in `exceptionHandleFuture`, - // it will be helpful to make the exception handling clearer. - exceptionHandleFuture(e.getCause()); - return null; - }, internalPinnedExecutor); + recoverTime.setRecoverStartTime(System.currentTimeMillis()); + pendingAckStore.replayAsync(this, internalPinnedExecutor); } + }).exceptionallyAsync(e -> { + exceptionHandleFuture(FutureUtil.unwrapCompletionException(e)); + return null; + }, internalPinnedExecutor); + try { + FutureUtil.completeAfter(storeFuture, pendingAckStoreProvider.newPendingAckStore(persistentSubscription)); + } catch (Exception e) { + storeFuture.completeExceptionally(e); } } @@ -983,8 +992,10 @@ public void completeHandleFuture() { } public void exceptionHandleFuture(Throwable t) { - if (isRetryableException(t)) { - this.state = State.None; + // Preserve Close atomically: closeAsync() changes the state outside the handle monitor. + // Resetting it after a separate close check could allow initPendingAckStore() to reopen the store. + boolean retryable = isRetryableException(t); + if (retryable && changeToNoneStateIfNotClosed()) { long retryTime = backoff.next().toMillis(); log.warn() .attr("name", persistentSubscription.getTopic().getName()) @@ -1001,8 +1012,8 @@ public void exceptionHandleFuture(Throwable t) { changeToErrorState(); // ToDo: Add a new serverError `TransactionComponentLoadFailedException` // and before that a `Unknown` will be returned first. - this.pendingAckStoreFuture = FutureUtil.failedFuture(new BrokerServiceException( - String.format("[%s][%s] Failed to init transaction pending ack.", topicName, subName))); + // Keep the original store future, including for non-retryable replay failures, so closeAsync() + // can still close the store. Report initialization failure through the handle future below. final boolean completedNow = this.pendingAckHandleCompletableFuture.completeExceptionally( new BrokerServiceException( String.format("[%s][%s] Failed to init transaction pending ack.", topicName, subName))); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java index 96653244b7642..210cc85f13840 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java @@ -59,8 +59,13 @@ protected void changeToCloseState() { STATE_UPDATER.set(this, State.Close); } + protected boolean changeToNoneStateIfNotClosed() { + return STATE_UPDATER.updateAndGet(this, current -> current == State.Close ? current : State.None) + != State.Close; + } + protected void changeToErrorState() { - STATE_UPDATER.set(this, State.Error); + STATE_UPDATER.updateAndGet(this, current -> current == State.Close ? current : State.Error); } public boolean checkIfReady() { 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..d72b00c81d4a3 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,29 +18,56 @@ */ package org.apache.pulsar.broker.transaction.pendingack.impl; +import static org.apache.pulsar.transaction.coordinator.impl.TxnLogBufferedWriter.BATCHED_ENTRY_DATA_PREFIX_MAGIC_NUMBER; +import static org.apache.pulsar.transaction.coordinator.impl.TxnLogBufferedWriter.BATCHED_ENTRY_DATA_PREFIX_VERSION; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; 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 io.netty.util.concurrent.DefaultThreadFactory; +import java.nio.ByteBuffer; 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.Future; 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.bookkeeper.mledger.impl.EntryImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.broker.transaction.TransactionTestBase; +import org.apache.pulsar.broker.transaction.pendingack.PendingAckReplyCallBack; +import org.apache.pulsar.broker.transaction.pendingack.PendingAckStore; +import org.apache.pulsar.broker.transaction.pendingack.TransactionPendingAckStoreProvider; import org.apache.pulsar.broker.transaction.util.LogIndexLagBackoff; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.common.api.proto.CommandAck; @@ -49,6 +76,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 +198,337 @@ public void testPendingAckStoreWithSlashSubscriptionName() throws Exception { closePendingAckStoreWithRetry(store); } + @Test + public void testReplayFailurePreservesClosedHandle() throws Exception { + PendingAckHandleImpl handle = new PendingAckHandleImpl(persistentSubscriptionMock); + handle.pendingAckHandleFuture().get(10, TimeUnit.SECONDS); + handle.closeAsync().get(10, TimeUnit.SECONDS); + + // Multiple late failures must not turn Close into Error and then enable a retry from Error. + for (int i = 0; i < 2; i++) { + handle.exceptionHandleFuture(new ManagedLedgerException.CursorAlreadyClosedException("closed")); + Assert.assertEquals(handle.getState(), PendingAckHandleState.State.Close); + Assert.assertFalse(handle.changeToNoneStateIfNotClosed()); + Assert.assertFalse(handle.changeToInitializingState()); + } + } + + @DataProvider(name = "lateReplayFailures") + public Object[][] lateReplayFailures() { + return new Object[][] { + {new ManagedLedgerException.CursorAlreadyClosedException("closed")}, + {new ManagedLedgerException.ManagedLedgerFencedException()} + }; + } + + @Test(dataProvider = "lateReplayFailures") + public void testReplayFailureDoesNotHideStoreFromConcurrentClose(ManagedLedgerException failure) throws Exception { + // Create the log so the handle opens a real store during initialization. + closePendingAckStoreWithRetry(createPendingAckStore(new TxnLogBufferedWriterConfig())); + PendingAckHandleImpl handle = new PendingAckHandleImpl(persistentSubscriptionMock); + handle.pendingAckHandleFuture().get(10, TimeUnit.SECONDS); + ManagedLedgerImpl ledger = (ManagedLedgerImpl) handle.getStoreManageLedger().get(10, TimeUnit.SECONDS); + ExecutorService closeExecutor = Executors.newSingleThreadExecutor( + new DefaultThreadFactory("pending-ack-close")); + try { + Future> closeResult; + synchronized (handle) { + closeResult = closeExecutor.submit(handle::closeAsync); + // closeAsync publishes Close before taking this monitor. Hold it as replayFailed does, + // ensuring the failure runs before closeAsync can retrieve the store future. + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(handle::checkIfClose); + handle.exceptionHandleFuture(failure); + } + closeResult.get(10, TimeUnit.SECONDS).get(10, TimeUnit.SECONDS); + Assert.assertEquals(ledger.getState(), ManagedLedgerImpl.State.Closed, + "The store must still be closed after a late replay failure"); + } finally { + closeExecutor.shutdownNow(); + ledger.close(); + } + } + + @Test + public void testCloseDuringPendingAckStoreCreation() throws Exception { + PersistentSubscription subscription = mock(PersistentSubscription.class, RETURNS_DEEP_STUBS); + doReturn(persistentSubscriptionMock.getTopicName()).when(subscription).getTopicName(); + when(subscription.getName()).thenReturn("close-during-store-creation"); + var pulsar = subscription.getTopic().getBrokerService().getPulsar(); + when(pulsar.getTransactionExecutorProvider().getExecutor(any(PendingAckHandleImpl.class))) + .thenReturn(internalPinnedExecutor); + TransactionPendingAckStoreProvider provider = mock(TransactionPendingAckStoreProvider.class); + when(pulsar.getTransactionPendingAckStoreProvider()).thenReturn(provider); + CompletableFuture initialized = new CompletableFuture<>(); + when(provider.checkInitializedBefore(subscription)).thenReturn(initialized); + PendingAckStore store = mock(PendingAckStore.class); + CompletableFuture storeClosed = new CompletableFuture<>(); + when(store.closeAsync()).thenReturn(storeClosed); + CompletableFuture storeCreated = new CompletableFuture<>(); + CompletableFuture> closeStarted = new CompletableFuture<>(); + PendingAckHandleImpl handle = new PendingAckHandleImpl(subscription); + when(provider.newPendingAckStore(subscription)).thenAnswer(invocation -> { + // Close after init's state check, but before the provider returns its future. + closeStarted.complete(handle.closeAsync()); + return storeCreated; + }); + try { + initialized.complete(true); + CompletableFuture closed = closeStarted.get(10, TimeUnit.SECONDS); + assertThat(closed).as("Close must wait for the store being created").isNotDone(); + storeCreated.complete(store); + verify(store, timeout(10000)).closeAsync(); + assertThat(closed).as("Close must wait for store cleanup").isNotDone(); + storeClosed.complete(null); + closed.get(10, TimeUnit.SECONDS); + handle.getInternalPinnedExecutor().submit(() -> { }).get(10, TimeUnit.SECONDS); + verify(store, never()).replayAsync(any(), any()); + assertThat(handle.getState()).isEqualTo(PendingAckHandleState.State.Close); + } finally { + storeCreated.complete(store); + storeClosed.complete(null); + handle.closeAsync().get(10, TimeUnit.SECONDS); + } + } + + @DataProvider(name = "replayEntryPayloads") + public Object[][] replayEntryPayloads() { + return new Object[][] { + {new byte[] {0}, true}, // Too short to decode the entry prefix. + {ByteBuffer.allocate(4).putShort(BATCHED_ENTRY_DATA_PREFIX_MAGIC_NUMBER) + .putShort(BATCHED_ENTRY_DATA_PREFIX_VERSION).array(), false} // Empty batch. + }; + } + + @Test(dataProvider = "replayEntryPayloads") + public void testReplayReleasesPolledAndQueuedEntries(byte[] payload, boolean expectFailure) { + ManagedCursor cursor = createReplayCursorMock(); + when(cursor.hasMoreEntries()).thenReturn(true); + EntryImpl first = EntryImpl.create(5, 10, payload); + EntryImpl second = EntryImpl.create(5, 11, payload); + doAnswer(invocation -> { + AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(1); + callback.readEntriesComplete(List.of(first, second), null); + return null; + }).when(cursor).asyncReadEntries(anyInt(), any(), any(), any()); + try { + MLPendingAckStore store = createPendingAckStoreForReplay(cursor); + PendingAckReplyCallBack callback = mock(PendingAckReplyCallBack.class); + store.new PendingAckReplay(callback).run(); + if (expectFailure) { + verify(callback).replayFailed(any(IndexOutOfBoundsException.class)); + verify(callback, never()).replayComplete(); + } else { + verify(callback).replayComplete(); + verify(callback, never()).replayFailed(any()); + } + Assert.assertEquals(first.refCnt(), 0, "The polled entry must be released"); + Assert.assertEquals(second.refCnt(), 0, "The queued entry must be released"); + } finally { + if (first.refCnt() > 0) { + first.release(); + } + if (second.refCnt() > 0) { + second.release(); + } + } + } + + /** + * 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.