[fix][broker] Fix multiple race conditions in InMemoryDelayedDeliveryTracker

Author: chamons

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

@@ -122,7 +122,7 @@ private static long trimLowerBit(long timestamp, int bits) {
     }
 
     @Override
-    public boolean addMessage(long ledgerId, long entryId, long deliverAt) {
+    public synchronized boolean addMessage(long ledgerId, long entryId, long deliverAt) {
         if (deliverAt < 0 || deliverAt <= getCutoffTime()) {
             messagesHaveFixedDelay = false;
             return false;
@@ -161,7 +161,7 @@ private void checkAndUpdateHighest(long deliverAt) {
      * Return true if there's at least a message that is scheduled to be delivered already.
      */
     @Override
-    public boolean hasMessageAvailable() {
+    public synchronized boolean hasMessageAvailable() {
         boolean hasMessageAvailable = !delayedMessageMap.isEmpty()
                 && delayedMessageMap.firstKey() <= getCutoffTime();
         if (!hasMessageAvailable) {
@@ -174,7 +174,7 @@ public boolean hasMessageAvailable() {
      * Get a set of position of messages that have already reached.
      */
     @Override
-    public NavigableSet<Position> getScheduledMessages(int maxMessages) {
+    public synchronized NavigableSet<Position> getScheduledMessages(int maxMessages) {
         int n = maxMessages;
         NavigableSet<Position> positions = new TreeSet<>();
         long cutoffTime = getCutoffTime();
@@ -237,7 +237,7 @@ public NavigableSet<Position> getScheduledMessages(int maxMessages) {
     }
 
     @Override
-    public CompletableFuture<Void> clear() {
+    public synchronized CompletableFuture<Void> clear() {
         this.delayedMessageMap.clear();
         this.delayedMessagesCount.set(0);
         return CompletableFuture.completedFuture(null);
@@ -262,7 +262,7 @@ public long getBufferMemoryUsage() {
     }
 
     @Override
-    public void close() {
+    public synchronized void close() {
         super.close();
     }
 
@@ -275,7 +275,7 @@ && getNumberOfDelayedMessages() >= fixedDelayDetectionLookahead
                 && !hasMessageAvailable();
     }
 
-    protected long nextDeliveryTime() {
+    protected synchronized long nextDeliveryTime() {
         return delayedMessageMap.firstKey();
     }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java

@@ -33,6 +33,12 @@
 import io.netty.util.concurrent.DefaultThreadFactory;
 import java.lang.reflect.Method;
 import java.time.Clock;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.NavigableMap;
 import java.util.Set;
 import java.util.TreeMap;
@@ -274,4 +280,100 @@ public void testDelaySequence(InMemoryDelayedDeliveryTracker tracker) throws Exc
         tracker.close();
     }
 
+    @Test(dataProvider = "delayedTracker")
+    public void testRaceConditionInUpdateTimer(InMemoryDelayedDeliveryTracker tracker) throws Exception {
+        final int numThreads = 15;
+        final int operationsPerThread = 2000;
+        final CountDownLatch startLatch = new CountDownLatch(1);
+        final CountDownLatch doneLatch = new CountDownLatch(numThreads);
+        final AtomicInteger errors = new AtomicInteger(0);
+        final AtomicReference<Exception> firstException = new AtomicReference<>();
+
+        @Cleanup("shutdown")
+        ExecutorService executorService = Executors.newFixedThreadPool(32);
+
+        for (int i = 0; i < numThreads / 4; i++) {
+            executorService.submit(() -> {
+                try {
+                    startLatch.await();
+                    for (int j = 0; j < operationsPerThread; j++) {
+                        tracker.getNumberOfDelayedMessages();
+                        Thread.sleep(1);
+                    }
+                } catch (Exception e) {
+                    errors.incrementAndGet();
+                    firstException.compareAndSet(null, e);
+                    e.printStackTrace();
+                } finally {
+                    doneLatch.countDown();
+                }
+            });
+        }
+
+        for (int i = numThreads / 4; i < numThreads; i++) {
+            executorService.submit(() -> {
+                try {
+                    startLatch.await();
+                    for (int j = 0; j < operationsPerThread; j++) {
+                        tracker.addMessage(1, 1, 10);
+                        Thread.sleep(1);
+                    }
+                } catch (Exception e) {
+                    errors.incrementAndGet();
+                    firstException.compareAndSet(null, e);
+                    e.printStackTrace();
+                } finally {
+                    doneLatch.countDown();
+                }
+            });
+        }
+
+        for (int i = numThreads / 4; i < numThreads; i++) {
+            executorService.submit(() -> {
+                try {
+                    startLatch.await();
+                    for (int j = 0; j < operationsPerThread; j++) {
+                        tracker.clear();
+                        Thread.sleep(1);
+                    }
+                } catch (Exception e) {
+                    errors.incrementAndGet();
+                    firstException.compareAndSet(null, e);
+                    e.printStackTrace();
+                } finally {
+                    doneLatch.countDown();
+                }
+            });
+        }
+
+        for (int i = numThreads / 4; i < numThreads; i++) {
+            executorService.submit(() -> {
+                try {
+                    startLatch.await();
+                    for (int j = 0; j < operationsPerThread; j++) {
+                        tracker.getScheduledMessages(1);
+                        Thread.sleep(1);
+                    }
+                } catch (Exception e) {
+                    errors.incrementAndGet();
+                    firstException.compareAndSet(null, e);
+                    e.printStackTrace();
+                } finally {
+                    doneLatch.countDown();
+                }
+            });
+        }
+
+        startLatch.countDown();
+        assertTrue(doneLatch.await(30, TimeUnit.SECONDS), "Test should complete within 30 seconds");
+
+        if (errors.get() > 0) {
+            Exception exception = firstException.get();
+            if (exception != null) {
+                System.err.println("First exception caught: " + exception.getMessage());
+                exception.printStackTrace();
+            }
+        }
+        assertEquals(errors.get(), 0, "No exceptions should occur during concurrent operations");
+    }
 }