[feat][broker] Implement topic-level delayed delivery tracking with in-memory manager

pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTrackerFactory.java

@@ -22,6 +22,9 @@
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.Timer;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import java.time.Clock;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
@@ -40,6 +43,34 @@ public class InMemoryDelayedDeliveryTrackerFactory implements DelayedDeliveryTra
 
     private long fixedDelayDetectionLookahead;
 
+    // Cache of topic-level managers: topic name -> manager instance
+    private final ConcurrentMap<String, TopicDelayedDeliveryTrackerManager> topicManagers = new ConcurrentHashMap<>();
+
+    public InMemoryDelayedDeliveryTrackerFactory() {
+
+    }
+
+    // Testing-friendly constructor and accessors
+    @VisibleForTesting
+    InMemoryDelayedDeliveryTrackerFactory(Timer timer, long tickTimeMillis,
+                                          boolean isDelayedDeliveryDeliverAtTimeStrict,
+                                          long fixedDelayDetectionLookahead) {
+        this.timer = timer;
+        this.tickTimeMillis = tickTimeMillis;
+        this.isDelayedDeliveryDeliverAtTimeStrict = isDelayedDeliveryDeliverAtTimeStrict;
+        this.fixedDelayDetectionLookahead = fixedDelayDetectionLookahead;
+    }
+
+    @VisibleForTesting
+    int getCachedManagersSize() {
+        return topicManagers.size();
+    }
+
+    @VisibleForTesting
+    boolean hasManagerForTopic(String topicName) {
+        return topicManagers.containsKey(topicName);
+    }
+
     @Override
     public void initialize(PulsarService pulsarService) {
         ServiceConfiguration config = pulsarService.getConfig();
@@ -54,7 +85,7 @@ public void initialize(PulsarService pulsarService) {
     public DelayedDeliveryTracker newTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
         String topicName = dispatcher.getTopic().getName();
         String subscriptionName = dispatcher.getSubscription().getName();
-        DelayedDeliveryTracker tracker =  DelayedDeliveryTracker.DISABLE;
+        DelayedDeliveryTracker tracker = DelayedDeliveryTracker.DISABLE;
         try {
             tracker = newTracker0(dispatcher);
         } catch (Exception e) {
@@ -66,13 +97,35 @@ public DelayedDeliveryTracker newTracker(AbstractPersistentDispatcherMultipleCon
     }
 
     @VisibleForTesting
-    InMemoryDelayedDeliveryTracker newTracker0(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
-        return new InMemoryDelayedDeliveryTracker(dispatcher, timer, tickTimeMillis,
-                isDelayedDeliveryDeliverAtTimeStrict, fixedDelayDetectionLookahead);
+    DelayedDeliveryTracker newTracker0(AbstractPersistentDispatcherMultipleConsumers dispatcher) {
+        String topicName = dispatcher.getTopic().getName();
+
+        // Get or create topic-level manager for this topic with onEmpty callback to remove from cache
+        final TopicDelayedDeliveryTrackerManager[] holder = new TopicDelayedDeliveryTrackerManager[1];
+        TopicDelayedDeliveryTrackerManager manager = topicManagers.computeIfAbsent(topicName, k -> {
+            InMemoryTopicDelayedDeliveryTrackerManager m = new InMemoryTopicDelayedDeliveryTrackerManager(
+                    timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict,
+                    fixedDelayDetectionLookahead, () -> topicManagers.remove(topicName, holder[0]));
+            holder[0] = m;
+            return m;
+        });
+
+        // Create a per-subscription view from the topic-level manager
+        return manager.createOrGetView(dispatcher);
     }
 
     @Override
     public void close() {
+        // Close all topic-level managers
+        for (TopicDelayedDeliveryTrackerManager manager : topicManagers.values()) {
+            try {
+                manager.close();
+            } catch (Exception e) {
+                log.warn("Failed to close topic-level delayed delivery manager", e);
+            }
+        }
+        topicManagers.clear();
+
         if (timer != null) {
             timer.stop();
         }