[fix][txn] Stop the pending ack replay loop from spinning forever

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<PendingAckMetadataEntry> 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<PendingAckMetadataEntry> 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<PendingAckMetadataEntry> deserializeEntry(Entry entry){
@@ -478,33 +514,89 @@ 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()
                     && outstandingReadsRequests.get() == 0) {
                 if (cursor.hasMoreEntries()) {
                     outstandingReadsRequests.incrementAndGet();
                     readAsync(NUMBER_OF_PER_READ_ENTRY, this);
+                } else if (entryQueue.size() == 0) {
+                    // Nothing left to read and everything read so far has been processed: the replay is
+                    // done. The loop condition in PendingAckReplay cannot detect this on its own because
+                    // it compares lastConfirmedEntry -- a snapshot taken when this store was created --
+                    // against currentLoadPosition, which starts at the cursor's mark-delete position,
+                    // while whether anything can still be read is decided by the cursor's read position.
+                    // Those two can disagree permanently, e.g. when the ledger holding the mark-delete
+                    // position was trimmed and the cursor was recovered onto a later ledger. Entries
+                    // below the mark-delete position have already been applied, so completing here is
+                    // correct. TopicTransactionBuffer's equivalent loop was fixed the same way in
+                    // https://github.com/apache/pulsar/pull/13739
+                    log.debug()
+                            .attr("lastConfirmedEntry", lastConfirmedEntry)
+                            .attr("currentLoadPosition", currentLoadPosition)
+                            .attr("markDeletePosition", cursor.getMarkDeletedPosition())
+                            .attr("readPosition", cursor.getReadPosition())
+                            .log("Pending ack replay stopped before reaching the last confirmed entry "
+                                    + "because the cursor has nothing left to read");
+                    isReadable = false;
                 }
             }
             return isReadable;
         }
 
         @Override
         public void readEntriesComplete(List<Entry> entries, Object ctx) {
-            entryQueue.fill(new MessagePassingQueue.Supplier<Entry>() {
-                private int i = 0;
-                @Override
-                public Entry get() {
-                    Entry entry = entries.get(i);
-                    i++;
-                    return entry;
+            List<Entry> entriesToRelease = null;
+            synchronized (stopLock) {
+                if (stopped) {
+                    // The replay already finished, so nothing will ever consume these.
+                    entriesToRelease = entries;
+                } else {
+                    int filled = entryQueue.fill(new MessagePassingQueue.Supplier<Entry>() {
+                        private int i = 0;
+                        @Override
+                        public Entry get() {
+                            Entry entry = entries.get(i);
+                            i++;
+                            return entry;
+                        }
+                    }, entries.size());
+                    if (filled < entries.size()) {
+                        entriesToRelease = entries.subList(filled, entries.size());
+                    }
                 }
-            }, entries.size());
-
+            }
+            // Released outside the lock: releasing an entry can run a deallocation callback.
+            if (entriesToRelease != null) {
+                entriesToRelease.forEach(Entry::release);
+            }
             outstandingReadsRequests.decrementAndGet();
         }
 
+        /**
+         * Stops accepting entries and returns everything still queued, so that the replay thread can
+         * release them. Must only be called by the replay thread, which is the queue's single consumer.
+         */
+        List<Entry> stopAndDrain() {
+            List<Entry> drained = new ArrayList<>();
+            synchronized (stopLock) {
+                stopped = true;
+                Entry entry;
+                while ((entry = entryQueue.poll()) != null) {
+                    drained.add(entry);
+                }
+            }
+            return drained;
+        }
+
         @Override
         public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
             if (managedLedger.getConfig().isAutoSkipNonRecoverableData()

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<PendingAckStore> 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)));

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() {