[feat][broker] Add configuration options for in-memory topic-level delayed delivery tracker and rename view class for clarity

Denovo1998 · Denovo1998 · commit 32f21700bf0b · 2025-11-02T12:13:35.000+08:00
diff --git a/conf/broker.conf b/conf/broker.conf
@@ -671,6 +671,24 @@ delayedDeliveryTickTimeMillis=1000
 # delayedDeliveryTickTimeMillis.
 isDelayedDeliveryDeliverAtTimeStrict=false
 
+# Minimum interval (in milliseconds) between prune attempts within the in-memory topic-level delayed delivery tracker.
+# (org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerFactory)
+# Set to a positive value to override the default adaptive interval based on delayedDeliveryTickTimeMillis.
+# Set to 0 or a negative value to use the default adaptive interval.
+delayedDeliveryPruneMinIntervalMillis=0
+
+# The ratio [0.0, 1.0] of subscriptions that need to be eligible for delivery in order to trigger an opportunistic
+# prune in the in-memory topic-level delayed delivery tracker.
+# (org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerFactory)
+# For example, 0.5 means prune when at least half of the subscriptions are eligible. Default is 0.5.
+delayedDeliveryPruneEligibleRatio=0.5
+
+# Idle timeout (in milliseconds) for the topic-level in-memory delayed delivery tracker manager.
+# (org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerFactory)
+# When the last subscription is unregistered, the manager will be removed from the factory cache after this idle
+# timeout, provided no new subscriptions have been registered in the meantime. Set to 0 to remove immediately (default).
+delayedDeliveryTopicManagerIdleMillis=0
+
 # The delayed message index bucket min index count.
 # When the index count of the current bucket is more than this value and all message indexes of current ledger
 # have already been added to the tracker we will seal the bucket.
diff --git a/conf/standalone.conf b/conf/standalone.conf
@@ -1439,6 +1439,24 @@ delayedDeliveryTickTimeMillis=1000
 # delayedDeliveryTickTimeMillis.
 isDelayedDeliveryDeliverAtTimeStrict=false
 
+# Minimum interval (in milliseconds) between prune attempts within the in-memory topic-level delayed delivery tracker.
+# (org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerFactory)
+# Set to a positive value to override the default adaptive interval based on delayedDeliveryTickTimeMillis.
+# Set to 0 or a negative value to use the default adaptive interval.
+delayedDeliveryPruneMinIntervalMillis=0
+
+# The ratio [0.0, 1.0] of subscriptions that need to be eligible for delivery in order to trigger an opportunistic
+# prune in the in-memory topic-level delayed delivery tracker.
+# (org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerFactory)
+# For example, 0.5 means prune when at least half of the subscriptions are eligible. Default is 0.5.
+delayedDeliveryPruneEligibleRatio=0.5
+
+# Idle timeout (in milliseconds) for the topic-level in-memory delayed delivery tracker manager.
+# (org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerFactory)
+# When the last subscription is unregistered, the manager will be removed from the factory cache after this idle
+# timeout, provided no new subscriptions have been registered in the meantime. Set to 0 to remove immediately (default).
+delayedDeliveryTopicManagerIdleMillis=0
+
 # The delayed message index bucket min index count.
 # When the index count of the current bucket is more than this value and all message indexes of current ledger
 # have already been added to the tracker we will seal the bucket.
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -421,6 +421,28 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
             + "logic to handle fixed delays in messages in a different way.")
     private long delayedDeliveryFixedDelayDetectionLookahead = 50_000;
 
+    @FieldContext(category = CATEGORY_SERVER, doc = """
+            Minimum interval (in milliseconds) between prune attempts within the in-memory topic-level delayed
+            delivery tracker. Set to a positive value to override the default adaptive interval based on
+            delayedDeliveryTickTimeMillis. Set to 0 or a negative value to use the default adaptive interval.
+            """)
+    private long delayedDeliveryPruneMinIntervalMillis = 0;
+
+    @FieldContext(category = CATEGORY_SERVER, doc = """
+            The ratio [0.0, 1.0] of subscriptions that need to be eligible for delivery in order to trigger an
+            opportunistic prune in the in-memory topic-level delayed delivery tracker. For example, 0.5 means prune
+            when at least half of the subscriptions are eligible. Default is 0.5.
+            """)
+    private double delayedDeliveryPruneEligibleRatio = 0.5;
+
+    @FieldContext(category = CATEGORY_SERVER, doc = """
+            Idle timeout (in milliseconds) for the topic-level in-memory delayed delivery tracker manager. When the
+            last subscription is unregistered, the manager will be removed from the factory cache after this idle
+            timeout, provided no new subscriptions have been registered in the meantime. Set to 0 to remove
+            immediately (default).
+            """)
+    private long delayedDeliveryTopicManagerIdleMillis = 0;
+
     @FieldContext(category = CATEGORY_SERVER, doc = """
             The max allowed delay for delayed delivery (in milliseconds). If the broker receives a message which \
             exceeds this max delay, then it will return an error to the producer. \
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryTopicDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryTopicDelayedDeliveryTracker.java
@@ -24,19 +24,18 @@
 import org.apache.bookkeeper.mledger.Position;
 
 /**
- * View object for a subscription that implements DelayedDeliveryTracker interface.
- * This view forwards all operations to the topic-level manager while maintaining
- * compatibility with existing dispatcher logic.
+ * Subscription-scoped tracker implementing {@link DelayedDeliveryTracker} by delegating all
+ * operations to the topic-level {@link InMemoryTopicDelayedDeliveryTrackerManager}.
  */
 @Slf4j
-public class InMemoryTopicDelayedDeliveryTrackerView implements DelayedDeliveryTracker {
+public class InMemoryTopicDelayedDeliveryTracker implements DelayedDeliveryTracker {
 
     private final InMemoryTopicDelayedDeliveryTrackerManager manager;
     private final InMemoryTopicDelayedDeliveryTrackerManager.SubContext subContext;
     private volatile boolean closed = false;
 
-    public InMemoryTopicDelayedDeliveryTrackerView(InMemoryTopicDelayedDeliveryTrackerManager manager,
-                                                   InMemoryTopicDelayedDeliveryTrackerManager.SubContext subContext) {
+    public InMemoryTopicDelayedDeliveryTracker(InMemoryTopicDelayedDeliveryTrackerManager manager,
+                                               InMemoryTopicDelayedDeliveryTrackerManager.SubContext subContext) {
         this.manager = manager;
         this.subContext = subContext;
     }
@@ -117,4 +116,4 @@ private void checkClosed() {
             throw new IllegalStateException("DelayedDeliveryTracker is already closed");
         }
     }
-}
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryTopicDelayedDeliveryTrackerFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryTopicDelayedDeliveryTrackerFactory.java
@@ -44,14 +44,14 @@ public class InMemoryTopicDelayedDeliveryTrackerFactory implements DelayedDelive
 
     private long fixedDelayDetectionLookahead;
 
+    // New tuning knobs
+    private long pruneMinIntervalMillis;
+    private double pruneEligibleRatio;
+    private long topicManagerIdleMillis;
+
     // Cache of topic-level managers: topic name -> manager instance
     private final ConcurrentMap<String, TopicDelayedDeliveryTrackerManager> topicManagers = new ConcurrentHashMap<>();
 
-    public InMemoryTopicDelayedDeliveryTrackerFactory() {
-
-    }
-
-    // Testing-friendly constructor and accessors
     @VisibleForTesting
     InMemoryTopicDelayedDeliveryTrackerFactory(Timer timer, long tickTimeMillis,
                                           boolean isDelayedDeliveryDeliverAtTimeStrict,
@@ -60,6 +60,9 @@ public InMemoryTopicDelayedDeliveryTrackerFactory() {
         this.tickTimeMillis = tickTimeMillis;
         this.isDelayedDeliveryDeliverAtTimeStrict = isDelayedDeliveryDeliverAtTimeStrict;
         this.fixedDelayDetectionLookahead = fixedDelayDetectionLookahead;
+        this.pruneMinIntervalMillis = 0;
+        this.pruneEligibleRatio = 0.5;
+        this.topicManagerIdleMillis = 0;
     }
 
     @VisibleForTesting
@@ -80,6 +83,9 @@ public void initialize(PulsarService pulsarService) {
         this.tickTimeMillis = config.getDelayedDeliveryTickTimeMillis();
         this.isDelayedDeliveryDeliverAtTimeStrict = config.isDelayedDeliveryDeliverAtTimeStrict();
         this.fixedDelayDetectionLookahead = config.getDelayedDeliveryFixedDelayDetectionLookahead();
+        this.pruneMinIntervalMillis = config.getDelayedDeliveryPruneMinIntervalMillis();
+        this.pruneEligibleRatio = config.getDelayedDeliveryPruneEligibleRatio();
+        this.topicManagerIdleMillis = config.getDelayedDeliveryTopicManagerIdleMillis();
     }
 
     @Override
@@ -106,13 +112,29 @@ DelayedDeliveryTracker newTracker0(AbstractPersistentDispatcherMultipleConsumers
         TopicDelayedDeliveryTrackerManager manager = topicManagers.computeIfAbsent(topicName, k -> {
             InMemoryTopicDelayedDeliveryTrackerManager m = new InMemoryTopicDelayedDeliveryTrackerManager(
                     timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict,
-                    fixedDelayDetectionLookahead, () -> topicManagers.remove(topicName, holder[0]));
+                    fixedDelayDetectionLookahead, pruneMinIntervalMillis, pruneEligibleRatio, () -> {
+                        if (topicManagerIdleMillis <= 0) {
+                            topicManagers.remove(topicName, holder[0]);
+                        } else {
+                            timer.newTimeout(__ -> {
+                                TopicDelayedDeliveryTrackerManager tm = holder[0];
+                                if (tm instanceof InMemoryTopicDelayedDeliveryTrackerManager) {
+                                    if (!((InMemoryTopicDelayedDeliveryTrackerManager) tm).hasActiveSubscriptions()) {
+                                        topicManagers.remove(topicName, tm);
+                                    }
+                                } else {
+                                    // If the manager has been replaced or removed, ensure entry is cleaned up
+                                    topicManagers.remove(topicName, tm);
+                                }
+                            }, topicManagerIdleMillis, TimeUnit.MILLISECONDS);
+                        }
+                    });
             holder[0] = m;
             return m;
         });
 
-        // Create a per-subscription view from the topic-level manager
-        return manager.createOrGetView(dispatcher);
+        // Create a per-subscription tracker from the topic-level manager
+        return manager.createOrGetTracker(dispatcher);
     }
 
     @Override
@@ -132,4 +154,3 @@ public void close() {
         }
     }
 }
-
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryTopicDelayedDeliveryTrackerManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryTopicDelayedDeliveryTrackerManager.java
@@ -82,6 +82,9 @@ public class InMemoryTopicDelayedDeliveryTrackerManager implements TopicDelayedD
     // Minimum interval between prunes
     private final long minPruneIntervalNanos;
 
+    // Ratio of eligible subscriptions required to opportunistically prune [0.0, 1.0]
+    private final double pruneEligibleRatio;
+
     // Fixed-delay detection (parity with legacy behavior)
     private final AtomicLong highestDeliveryTimeTracked = new AtomicLong(0);
     private volatile boolean messagesHaveFixedDelay = true;
@@ -131,23 +134,32 @@ long getCutoffTime() {
     public InMemoryTopicDelayedDeliveryTrackerManager(Timer timer, long tickTimeMillis, Clock clock,
                                                       boolean isDelayedDeliveryDeliverAtTimeStrict,
                                                       long fixedDelayDetectionLookahead) {
-        this(timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict, fixedDelayDetectionLookahead, null);
+        this(timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict, fixedDelayDetectionLookahead,
+                0, 0.5, null);
     }
 
     public InMemoryTopicDelayedDeliveryTrackerManager(Timer timer, long tickTimeMillis, Clock clock,
                                                       boolean isDelayedDeliveryDeliverAtTimeStrict,
                                                       long fixedDelayDetectionLookahead,
+                                                      long pruneMinIntervalMillis,
+                                                      double pruneEligibleRatio,
                                                       Runnable onEmptyCallback) {
         this.timer = timer;
         this.tickTimeMillis = tickTimeMillis;
         this.clock = clock;
         this.isDelayedDeliveryDeliverAtTimeStrict = isDelayedDeliveryDeliverAtTimeStrict;
         this.fixedDelayDetectionLookahead = fixedDelayDetectionLookahead;
         this.onEmptyCallback = onEmptyCallback;
-        // Default prune throttle interval: clamp to [5ms, 50ms] using tickTimeMillis as hint
-        // TODO: make configurable if needed
-        long pruneMs = Math.max(5L, Math.min(50L, tickTimeMillis));
+        // Prune throttle interval: use configured override if positive, otherwise adaptive clamp [5ms, 50ms]
+        long pruneMs = pruneMinIntervalMillis > 0
+                ? pruneMinIntervalMillis
+                : Math.max(5L, Math.min(50L, tickTimeMillis));
         this.minPruneIntervalNanos = TimeUnit.MILLISECONDS.toNanos(pruneMs);
+        // Prune eligible ratio: clamp into [0.0, 1.0]
+        if (Double.isNaN(pruneEligibleRatio)) {
+            pruneEligibleRatio = 0.5;
+        }
+        this.pruneEligibleRatio = Math.max(0.0, Math.min(1.0, pruneEligibleRatio));
     }
 
     // We bucket messages by aligning the deliverAt timestamp to the start of the logical tick window:
@@ -167,13 +179,13 @@ private long bucketStart(long timestamp) {
     }
 
     @Override
-    public DelayedDeliveryTracker createOrGetView(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
+    public DelayedDeliveryTracker createOrGetTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
         String subscriptionName = dispatcher.getSubscription().getName();
 
         SubContext subContext = subscriptionContexts.computeIfAbsent(subscriptionName,
                 k -> new SubContext(dispatcher, tickTimeMillis, isDelayedDeliveryDeliverAtTimeStrict,
                         fixedDelayDetectionLookahead, clock));
-        return new InMemoryTopicDelayedDeliveryTrackerView(this, subContext);
+        return new InMemoryTopicDelayedDeliveryTracker(this, subContext);
     }
 
     @Override
@@ -209,6 +221,13 @@ public void unregister(AbstractPersistentDispatcherMultipleConsumers dispatcher)
         }
     }
 
+    /**
+     * Whether there are active subscriptions registered with this manager.
+     */
+    public boolean hasActiveSubscriptions() {
+        return !subscriptionContexts.isEmpty();
+    }
+
     @Override
     public void onTickTimeUpdated(long newTickTimeMillis) {
         if (this.tickTimeMillis == newTickTimeMillis) {
@@ -560,7 +579,7 @@ public void run(Timeout timeout) throws Exception {
             // If a significant portion of subscriptions are eligible, opportunistically prune (throttled)
             int subs = subscriptionContexts.size();
             int eligible = toTrigger.size();
-            int threshold = Math.max(1, subs / 2);
+            int threshold = Math.max(1, (int) Math.ceil(subs * pruneEligibleRatio));
             if (eligible >= threshold) {
                 maybePruneByTime();
             }
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/TopicDelayedDeliveryTrackerManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/TopicDelayedDeliveryTrackerManager.java
@@ -26,18 +26,18 @@
  * allowing different implementations (in-memory, bucket-based) to share the same contract.
  * <p>
  * The manager maintains a single global delayed message index per topic that is shared by all
- * subscriptions, and provides per-subscription "view" objects that implement DelayedDeliveryTracker
+ * subscriptions, and provides per-subscription tracker objects that implement DelayedDeliveryTracker
  * interface for compatibility with existing dispatcher logic.
  */
 public interface TopicDelayedDeliveryTrackerManager extends AutoCloseable {
 
     /**
-     * Create or get a delayed delivery tracker view for the specified subscription.
+     * Create or get a delayed delivery tracker for the specified subscription.
      *
      * @param dispatcher the dispatcher instance for the subscription
-     * @return a DelayedDeliveryTracker view for the subscription
+     * @return a DelayedDeliveryTracker bound to the subscription
      */
-    DelayedDeliveryTracker createOrGetView(AbstractPersistentDispatcherMultipleConsumers dispatcher);
+    DelayedDeliveryTracker createOrGetTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher);
 
     /**
      * Unregister a subscription from the manager.
@@ -71,4 +71,4 @@ public interface TopicDelayedDeliveryTrackerManager extends AutoCloseable {
      * Close the manager and release all resources.
      */
     void close();
-}
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -53,7 +53,7 @@
 import org.apache.pulsar.broker.delayed.DelayedDeliveryTracker;
 import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerFactory;
 import org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker;
-import org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerView;
+import org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTracker;
 import org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker;
 import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
 import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
@@ -1485,7 +1485,7 @@ public void markDeletePositionMoveForward() {
             trackerOpt = this.delayedDeliveryTracker;
         }
         trackerOpt.ifPresent(tracker -> {
-            if (tracker instanceof InMemoryTopicDelayedDeliveryTrackerView view) {
+            if (tracker instanceof InMemoryTopicDelayedDeliveryTracker view) {
                 Position md = cursor.getMarkDeletedPosition();
                 if (md != null) {
                     // Only update the cached mark-delete; avoid heavy operations in this callback
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java
@@ -50,7 +50,7 @@
 import org.apache.pulsar.broker.delayed.DelayedDeliveryTracker;
 import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerFactory;
 import org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker;
-import org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTrackerView;
+import org.apache.pulsar.broker.delayed.InMemoryTopicDelayedDeliveryTracker;
 import org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker;
 import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
 import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers;
@@ -1316,7 +1316,7 @@ public void markDeletePositionMoveForward() {
             trackerOpt = this.delayedDeliveryTracker;
         }
         trackerOpt.ifPresent(tracker -> {
-            if (tracker instanceof InMemoryTopicDelayedDeliveryTrackerView view) {
+            if (tracker instanceof InMemoryTopicDelayedDeliveryTracker view) {
                 Position md = cursor.getMarkDeletedPosition();
                 if (md != null) {
                     // Only update the cached mark-delete; avoid heavy operations in this callback
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryTopicDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryTopicDeliveryTrackerTest.java
