(pulsar) 10/18: [fix][broker] Run the message expiry check off the topic policy update path (#26040)

[lh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit 0ac71ff034fd8adc67ea29ceb406a78863d058d4
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jun 18 02:09:04 2026 +0300

    [fix][broker] Run the message expiry check off the topic policy update path (#26040)
    
    (cherry picked from commit 476312837b5660b23b5ea2c6ad112bdfba88f2ca)
---
 .../pulsar/broker/service/BrokerService.java       |  1 +
 .../broker/service/persistent/PersistentTopic.java | 26 ++++++++++++++++++++--
 .../pulsar/broker/service/MessageTTLTest.java      |  7 +++---
 3 files changed, 29 insertions(+), 5 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 0664d9ff290b..275d0a61a89f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -285,6 +285,7 @@ public class BrokerService implements Closeable {
             .register();
 
     private final SingleThreadNonConcurrentFixedRateScheduler inactivityMonitor;
+    @Getter
     private final SingleThreadNonConcurrentFixedRateScheduler messageExpiryMonitor;
     private final SingleThreadNonConcurrentFixedRateScheduler compactionMonitor;
     private final SingleThreadNonConcurrentFixedRateScheduler consumedLedgersMonitor;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 4dd2abf9ef8b..6e85477a1da6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -226,6 +226,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
     private final TopicName shadowSourceTopic;
 
     public static final String DEDUPLICATION_CURSOR_NAME = "pulsar.dedup";
+    private volatile int lastMessageTtlInSeconds;
 
     public static boolean isDedupCursorName(String name) {
         return DEDUPLICATION_CURSOR_NAME.equals(name);
@@ -3719,8 +3720,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
         });
         producers.values().forEach(producer -> applyPoliciesFutureList.add(
                 producer.checkPermissionsAsync().thenRun(producer::checkEncryption)));
-        // Check message expiry.
-        applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> checkMessageExpiry()));
+
+        // Check message expiry in the background so that it doesn't block updating or loading topic policies.
+        maybeCheckMessageExpiryOnPolicyUpdateInBackground();
 
         // Update rate limiters.
         applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> updateDispatchRateLimiter()));
@@ -3745,6 +3747,26 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
         return applyPoliciesFutureList;
     }
 
+    /**
+     * Triggers a message-expiry check on a policy update, but only when the message TTL actually changed, and
+     * runs it on the messageExpiryMonitor thread (the same thread the periodic expiry check uses) so it never
+     * blocks updating or loading topic policies. Unlike before, the expiry check is no longer part of the
+     * returned {@link #applyUpdatedTopicPolicies()} futures, so callers no longer wait for it; the periodic
+     * monitor provides eventual coverage and the check is idempotent.
+     */
+    private void maybeCheckMessageExpiryOnPolicyUpdateInBackground() {
+        int messageTtlInSeconds = topicPolicies.getMessageTTLInSeconds().get();
+        // don't check if the message expiry hasn't changed
+        if (messageTtlInSeconds > 0) {
+            if (lastMessageTtlInSeconds != messageTtlInSeconds) {
+                lastMessageTtlInSeconds = messageTtlInSeconds;
+                brokerService.getMessageExpiryMonitor().execute(this::checkMessageExpiry);
+            }
+        } else {
+            lastMessageTtlInSeconds = messageTtlInSeconds;
+        }
+    }
+
     /**
      *
      * @return Backlog quota for topic
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java
index beb054d86f5e..699f5638aabd 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/MessageTTLTest.java
@@ -19,7 +19,7 @@
 package org.apache.pulsar.broker.service;
 
 import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
@@ -138,12 +138,13 @@ public class MessageTTLTest extends BrokerTestBase {
         Policies policies = admin.namespaces().getPolicies(namespace);
         policies.message_ttl_in_seconds = 10;
         topicRefMock.onPoliciesUpdate(policies);
-        verify(topicRefMock, times(1)).checkMessageExpiry();
+        // checkMessageExpiry now runs asynchronously on the messageExpiryMonitor thread, so wait for it.
+        verify(topicRefMock, timeout(5000).times(1)).checkMessageExpiry();
 
         TopicPolicies topicPolicies = new TopicPolicies();
         topicPolicies.setMessageTTLInSeconds(5);
         topicRefMock.onUpdate(topicPolicies);
-        verify(topicRefMock, times(2)).checkMessageExpiry();
+        verify(topicRefMock, timeout(5000).times(2)).checkMessageExpiry();
     }
 
     @Test