[fix][broker] Prevent backlog quota cursor updates during topic close

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java

@@ -23,6 +23,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import lombok.CustomLog;
 import lombok.Getter;
@@ -95,10 +96,8 @@ public BacklogQuotaImpl getBacklogQuota(NamespaceName namespace, BacklogQuotaTyp
     public void handleExceededBacklogQuota(PersistentTopic persistentTopic, BacklogQuotaType backlogQuotaType,
                                            boolean preciseTimeBasedBacklogQuotaCheck) {
         if (persistentTopic.isFenced() || persistentTopic.isClosingOrDeleting()) {
-            // Skip eviction work on a topic that is being torn down or transiently fenced.
-            // Mutating cursors here (skipEntries / markDeletePosition) contends with the
-            // delete path and can keep namespace force-delete from completing in time;
-            // the entries are about to be discarded anyway.
+            // Skip quota handling on a topic that is temporarily unavailable or being torn down.
+            // For close/delete, mutating cursors here can contend with the teardown path.
             log.debug()
                     .attr("topic", persistentTopic.getName())
                     .attr("backlogQuotaType", backlogQuotaType)
@@ -196,15 +195,21 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
                     log.debug().attr("slowestConsumer", slowestConsumer).log("no messages to skip for");
                     break;
                 }
-                // Skip messages on the slowest consumer
-                log.debug()
-                        .attr("topic", persistentTopic.getName())
-                        .attr("messagesToSkip", messagesToSkip)
-                        .attr("consumer", slowestConsumer.getName())
-                        .attr("entriesInBacklog", entriesInBacklog)
-                        .log("Skipping messages on slowest consumer having backlog entries");
-                slowestConsumer.skipEntries(messagesToSkip, IndividualDeletedEntries.Include);
-                markDeletePositionMoveForward(persistentTopic, slowestConsumer);
+                beforeBacklogQuotaCursorMutation(persistentTopic);
+                if (!runCursorMutationIfTopicNotClosingOrDeleting(persistentTopic, () -> {
+                    // Skip messages on the slowest consumer
+                    log.debug()
+                            .attr("topic", persistentTopic.getName())
+                            .attr("messagesToSkip", messagesToSkip)
+                            .attr("consumer", slowestConsumer.getName())
+                            .attr("entriesInBacklog", entriesInBacklog)
+                            .log("Skipping messages on slowest consumer having backlog entries");
+                    slowestConsumer.skipEntries(messagesToSkip, IndividualDeletedEntries.Include);
+                    markDeletePositionMoveForward(persistentTopic, slowestConsumer);
+                    return null;
+                })) {
+                    break;
+                }
             } catch (Exception e) {
                 log.error()
                         .attr("topic", persistentTopic.getName())
@@ -267,8 +272,14 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
                     if (ledgerInfo == null) {
                         long ledgerId = mLedger.getLedgersInfo().ceilingKey(oldestPosition.getLedgerId() + 1);
                         Position nextPosition = PositionFactory.create(ledgerId, -1);
-                        slowestConsumer.markDelete(nextPosition);
-                        markDeletePositionMoveForward(persistentTopic, slowestConsumer);
+                        beforeBacklogQuotaCursorMutation(persistentTopic);
+                        if (!runCursorMutationIfTopicNotClosingOrDeleting(persistentTopic, () -> {
+                            slowestConsumer.markDelete(nextPosition);
+                            markDeletePositionMoveForward(persistentTopic, slowestConsumer);
+                            return null;
+                        })) {
+                            break;
+                        }
                         continue;
                     }
                     // Timestamp only > 0 if ledger has been closed
@@ -278,8 +289,14 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
                         long ledgerId = mLedger.getLedgersInfo().ceilingKey(oldestPosition.getLedgerId() + 1);
                         Position nextPosition = PositionFactory.create(ledgerId, -1);
                         if (!nextPosition.equals(oldestPosition)) {
-                            slowestConsumer.markDelete(nextPosition);
-                            markDeletePositionMoveForward(persistentTopic, slowestConsumer);
+                            beforeBacklogQuotaCursorMutation(persistentTopic);
+                            if (!runCursorMutationIfTopicNotClosingOrDeleting(persistentTopic, () -> {
+                                slowestConsumer.markDelete(nextPosition);
+                                markDeletePositionMoveForward(persistentTopic, slowestConsumer);
+                                return null;
+                            })) {
+                                break;
+                            }
                             continue;
                         }
                     }
@@ -366,6 +383,21 @@ private void markDeletePositionMoveForward(PersistentTopic persistentTopic, Mana
         }
     }
 
+    @VisibleForTesting
+    protected void beforeBacklogQuotaCursorMutation(PersistentTopic persistentTopic) {
+        // No-op.
+    }
+
+    private boolean runCursorMutationIfTopicNotClosingOrDeleting(PersistentTopic persistentTopic,
+                                                                 Callable<Void> mutation) throws Exception {
+        boolean didRun = persistentTopic.runWithTopicCloseReadLock(mutation);
+        if (!didRun) {
+            log.debug()
+                    .attr("topic", persistentTopic.getName())
+                    .log("Stopping backlog-quota eviction because topic is closing or deleting");
+        }
+        return didRun;
+    }
 
     /**
      * Compute the target value after backlog eviction.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java

@@ -23,6 +23,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.SortedMap;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -93,6 +94,11 @@ public boolean isAutoSkipNonRecoverableData() {
                 && this.cursor.getManagedLedger().getConfig().isAutoSkipNonRecoverableData();
     }
 
+    @VisibleForTesting
+    public boolean isExpirationCheckInProgress() {
+        return expirationCheckInProgress == TRUE;
+    }
+
     @Override
     public CompletableFuture<Boolean> expireMessagesAsync(int messageTTLInSeconds) {
         return CompletableFuture.supplyAsync(() -> expireMessages(messageTTLInSeconds), topic.getOrderedExecutor());
@@ -244,10 +250,17 @@ public void findEntryComplete(Position position, Object ctx) {
             log.info()
                     .attr("position", position)
                     .log("Expiring all messages until position");
-            Position prevMarkDeletePos = cursor.getMarkDeletedPosition();
-            cursor.asyncMarkDelete(position, null, markDeleteCallback, cursor.getNumberOfEntriesInBacklog(false));
-            if (!Objects.equals(cursor.getMarkDeletedPosition(), prevMarkDeletePos) && subscription != null) {
-                subscription.updateLastMarkDeleteAdvancedTimestamp();
+            beforeCursorMarkDelete(position);
+            if (!runCursorMarkDeleteIfTopicNotClosingOrDeleting(position, () -> {
+                Position prevMarkDeletePos = cursor.getMarkDeletedPosition();
+                cursor.asyncMarkDelete(position, null, markDeleteCallback, cursor.getNumberOfEntriesInBacklog(false));
+                if (!Objects.equals(cursor.getMarkDeletedPosition(), prevMarkDeletePos) && subscription != null) {
+                    subscription.updateLastMarkDeleteAdvancedTimestamp();
+                }
+                return null;
+            })) {
+                expirationCheckInProgress = FALSE;
+                updateRates();
             }
         } else {
             log.debug("No messages to expire");
@@ -256,6 +269,29 @@ public void findEntryComplete(Position position, Object ctx) {
         }
     }
 
+    @VisibleForTesting
+    protected void beforeCursorMarkDelete(Position position) {
+        // No-op.
+    }
+
+    private boolean runCursorMarkDeleteIfTopicNotClosingOrDeleting(Position position, Callable<Void> markDelete) {
+        try {
+            boolean didRun = topic.runWithTopicCloseReadLock(markDelete);
+            if (!didRun) {
+                log.debug()
+                        .attr("position", position)
+                        .log("Skipping message expiry mark-delete because topic is closing or deleting");
+            }
+            return didRun;
+        } catch (Exception e) {
+            log.warn()
+                    .attr("position", position)
+                    .exception(e)
+                    .log("Message expiry failed - mark delete failed");
+            return false;
+        }
+    }
+
     @Override
     public void findEntryFailed(ManagedLedgerException exception, Optional<Position> failedReadPosition, Object ctx) {
         log.debug()

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -43,6 +43,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -1749,6 +1750,26 @@ public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect
         return close(true, closeWithoutWaitingClientDisconnect);
     }
 
+    /**
+     * Executes an action while holding the read side of the topic close/delete lock.
+     * Topic close and delete acquire the write side before setting {@link #isClosingOrDeleting}, so they cannot start
+     * between the state check and the action.
+     *
+     * @return true if the action ran, or false if close/delete had already started
+     */
+    public boolean runWithTopicCloseReadLock(Callable<Void> action) throws Exception {
+        lock.readLock().lock();
+        try {
+            if (isClosingOrDeleting) {
+                return false;
+            }
+            action.call();
+            return true;
+        } finally {
+            lock.readLock().unlock();
+        }
+    }
+
     private enum CloseTypes {
         transferring,
         notWaitDisconnectClients,

pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/PersistentMessageExpiryMonitorMockTest.java

@@ -26,6 +26,7 @@
 import static org.testng.Assert.assertTrue;
 import java.util.NavigableMap;
 import java.util.TreeMap;
+import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -52,7 +53,7 @@ public class PersistentMessageExpiryMonitorMockTest {
     private BrokerService mockBrokerService;
 
     @BeforeMethod
-    public void setup() {
+    public void setup() throws Exception {
         mockTopic = mock(PersistentTopic.class);
         mockCursor = mock(ManagedCursor.class);
         mockManagedLedger = mock(ManagedLedger.class);
@@ -66,6 +67,11 @@ public void setup() {
         ServiceConfiguration config = new ServiceConfiguration();
         when(mockBrokerService.pulsar()).thenReturn(mockPulsarService);
         when(mockPulsarService.getConfig()).thenReturn(config);
+        doAnswer(invocation -> {
+            Callable<Void> action = invocation.getArgument(0);
+            action.call();
+            return true;
+        }).when(mockTopic).runWithTopicCloseReadLock(any());
     }
 
     /**

pulsar-broker/src/test/java/org/apache/bookkeeper/mledger/impl/PersistentMessageExpiryMonitorTest.java

@@ -27,13 +27,15 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.CustomLog;
 import org.apache.bookkeeper.client.api.ReadHandle;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.pulsar.broker.BrokerTestUtil;
 import org.apache.pulsar.broker.service.persistent.PersistentMessageExpiryMonitor;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
@@ -198,4 +200,70 @@ void testTopicExpireMessages() throws Exception {
         producer.close();
         admin.topics().delete(topicName);
     }
+
+    @Test
+    void testExpireMessagesRaceWithTopicCloseDoesNotMarkDelete() throws Exception {
+        final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp-closing-expiry");
+        final String subName = "closing-expiry-sub";
+        admin.topics().createNonPartitionedTopic(topicName);
+        admin.topics().createSubscription(topicName, subName, MessageId.earliest);
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+        for (int i = 0; i < 3; i++) {
+            producer.send("message-" + i);
+        }
+        producer.close();
+
+        PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
+        PersistentSubscription subscription = topic.getSubscription(subName);
+        ManagedCursorImpl cursor = (ManagedCursorImpl) subscription.getCursor();
+        Position markDeleteBeforeExpiry = cursor.getMarkDeletedPosition();
+
+        Thread.sleep(TimeUnit.SECONDS.toMillis(2));
+        BlockingPersistentMessageExpiryMonitor monitor =
+                new BlockingPersistentMessageExpiryMonitor(topic, subName, cursor, subscription);
+        CompletableFuture<Boolean> expireFuture = CompletableFuture.supplyAsync(() -> monitor.expireMessages(1));
+        monitor.awaitBeforeCursorMarkDelete();
+
+        CompletableFuture<Void> closeFuture = topic.close(false);
+        Awaitility.await().untilAsserted(() -> Assert.assertTrue(topic.isClosingOrDeleting()));
+        monitor.allowCursorMarkDelete();
+
+        Assert.assertTrue(expireFuture.get(30, TimeUnit.SECONDS));
+        Awaitility.await().untilAsserted(() -> Assert.assertFalse(monitor.isExpirationCheckInProgress()));
+        assertEquals(cursor.getMarkDeletedPosition(), markDeleteBeforeExpiry);
+        closeFuture.get(30, TimeUnit.SECONDS);
+    }
+
+    private static class BlockingPersistentMessageExpiryMonitor extends PersistentMessageExpiryMonitor {
+        private final CountDownLatch beforeCursorMarkDelete = new CountDownLatch(1);
+        private final CountDownLatch continueCursorMarkDelete = new CountDownLatch(1);
+        private final AtomicBoolean blocked = new AtomicBoolean();
+
+        BlockingPersistentMessageExpiryMonitor(PersistentTopic topic, String subscriptionName, ManagedCursorImpl cursor,
+                                               PersistentSubscription subscription) {
+            super(topic, subscriptionName, cursor, subscription);
+        }
+
+        @Override
+        protected void beforeCursorMarkDelete(Position position) {
+            if (!blocked.compareAndSet(false, true)) {
+                return;
+            }
+            beforeCursorMarkDelete.countDown();
+            try {
+                Assert.assertTrue(continueCursorMarkDelete.await(30, TimeUnit.SECONDS));
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new RuntimeException(e);
+            }
+        }
+
+        void awaitBeforeCursorMarkDelete() throws InterruptedException {
+            Assert.assertTrue(beforeCursorMarkDelete.await(30, TimeUnit.SECONDS));
+        }
+
+        void allowCursorMarkDelete() {
+            continueCursorMarkDelete.countDown();
+        }
+    }
 }