[feat][broker] Replace highestDeliveryTimeTracked with AtomicLong for thread-safe updates and improve concurrency handling in InMemoryTopicDelayedDeliveryTrackerManager

Author: Denovo1998

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

@@ -83,7 +83,7 @@ public class InMemoryTopicDelayedDeliveryTrackerManager implements TopicDelayedD
     private final long minPruneIntervalNanos;
 
     // Fixed-delay detection (parity with legacy behavior)
-    private volatile long highestDeliveryTimeTracked = 0;
+    private final AtomicLong highestDeliveryTimeTracked = new AtomicLong(0);
     private volatile boolean messagesHaveFixedDelay = true;
 
     // Per-bucket locks (timestamp -> lock) for fine-grained concurrency
@@ -302,10 +302,13 @@ boolean addMessageForSub(SubContext subContext, long ledgerId, long entryId, lon
     }
 
     private void checkAndUpdateHighest(long deliverAt) {
-        if (deliverAt < (highestDeliveryTimeTracked - tickTimeMillis)) {
-            messagesHaveFixedDelay = false;
-        }
-        highestDeliveryTimeTracked = Math.max(highestDeliveryTimeTracked, deliverAt);
+        long current;
+        do {
+            current = highestDeliveryTimeTracked.get();
+            if (deliverAt < (current - tickTimeMillis)) {
+                messagesHaveFixedDelay = false;
+            }
+        } while (deliverAt > current && !highestDeliveryTimeTracked.compareAndSet(current, deliverAt));
     }
 
     /**

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

@@ -33,7 +33,7 @@ public class InMemoryTopicDelayedDeliveryTrackerView implements DelayedDeliveryT
 
     private final InMemoryTopicDelayedDeliveryTrackerManager manager;
     private final InMemoryTopicDelayedDeliveryTrackerManager.SubContext subContext;
-    private boolean closed = false;
+    private volatile boolean closed = false;
 
     public InMemoryTopicDelayedDeliveryTrackerView(InMemoryTopicDelayedDeliveryTrackerManager manager,
                                                    InMemoryTopicDelayedDeliveryTrackerManager.SubContext subContext) {