[improve][broker]Improve the anti-concurrency mechanism expirationCheckInProgress

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java

@@ -193,6 +193,7 @@ public void markDeleteComplete(Object ctx) {
             if (subscription != null && subscription.getType() == SubType.Key_Shared) {
                 subscription.getDispatcher().markDeletePositionMoveForward();
             }
+            expirationCheckInProgress = FALSE;
             if (log.isDebugEnabled()) {
                 log.debug("[{}][{}] Mark deleted {} messages", topicName, subName, numMessagesExpired);
             }
@@ -201,6 +202,7 @@ public void markDeleteComplete(Object ctx) {
         @Override
         public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
             log.warn("[{}][{}] Message expiry failed - mark delete failed", topicName, subName, exception);
+            expirationCheckInProgress = FALSE;
             updateRates();
         }
     };
@@ -219,9 +221,9 @@ public void findEntryComplete(Position position, Object ctx) {
             if (log.isDebugEnabled()) {
                 log.debug("[{}][{}] No messages to expire", topicName, subName);
             }
+            expirationCheckInProgress = FALSE;
             updateRates();
         }
-        expirationCheckInProgress = FALSE;
     }
 
     @Override

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java

@@ -19,13 +19,10 @@
 package org.apache.pulsar.broker.service;
 
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
@@ -44,6 +41,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.ws.rs.client.Entity;
 import javax.ws.rs.core.MediaType;
@@ -698,18 +696,24 @@ void testMessageExpiryWithPosition() throws Exception {
         }
         when(topic.getLastPosition()).thenReturn(positions.get(positions.size() - 1));
 
-        PersistentMessageExpiryMonitor monitor = spy(new PersistentMessageExpiryMonitor(topic,
-                cursor.getName(), cursor, subscription));
+        AtomicInteger counterCalledFindEntryComplete = new AtomicInteger(0);
+        PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(topic,
+                cursor.getName(), cursor, subscription) {
+            @Override
+            public void findEntryComplete(Position position, Object ctx) {
+                super.findEntryComplete(position, ctx);
+                counterCalledFindEntryComplete.incrementAndGet();
+            }
+        };
         assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(0).getLedgerId(), -1));
         boolean issued;
 
         // Expire by position and verify mark delete position of cursor.
         issued = monitor.expireMessages(positions.get(15));
-        Awaitility.await().untilAsserted(() -> verify(monitor, times(1)).findEntryComplete(any(), any()));
+        Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 1));
         assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(15).getLedgerId(),
                 positions.get(15).getEntryId()));
         assertTrue(issued);
-        clearInvocations(monitor);
 
         // Expire by position beyond last position and nothing should happen.
         issued = monitor.expireMessages(PositionFactory.create(100, 100));
@@ -719,32 +723,26 @@ void testMessageExpiryWithPosition() throws Exception {
 
         // Expire by position again and verify mark delete position of cursor didn't change.
         issued = monitor.expireMessages(positions.get(15));
-        Awaitility.await().untilAsserted(() -> verify(monitor,
-                times(1)).findEntryComplete(any(), any()));
+        Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 2));
         assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(15).getLedgerId(),
                 positions.get(15).getEntryId()));
         assertTrue(issued);
-        clearInvocations(monitor);
 
         // Expire by position before current mark delete position and
         // verify mark delete position of cursor didn't change.
         issued = monitor.expireMessages(positions.get(10));
-        Awaitility.await().untilAsserted(() -> verify(monitor,
-                times(1)).findEntryComplete(any(), any()));
+        Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 3));
         assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(15).getLedgerId(),
                 positions.get(15).getEntryId()));
         assertTrue(issued);
-        clearInvocations(monitor);
 
         // Expire by position after current mark delete position and
         // verify mark delete position of cursor move to new position.
         issued = monitor.expireMessages(positions.get(16));
-        Awaitility.await().untilAsserted(() -> verify(monitor,
-                times(1)).findEntryComplete(any(), any()));
+        Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 4));
         assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(16).getLedgerId(),
                 positions.get(16).getEntryId()));
         assertTrue(issued);
-        clearInvocations(monitor);
 
         ManagedCursorImpl mockCursor = mock(ManagedCursorImpl.class);
         doReturn("cursor").when(mockCursor).getName();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitorTest.java

@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doAnswer;
+import static org.testng.AssertJUnit.assertEquals;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.awaitility.Awaitility;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class PersistentMessageExpiryMonitorTest extends ProducerConsumerBase {
+
+    @BeforeClass(alwaysRun = true)
+    @Override
+    protected void setup() throws Exception {
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @AfterClass(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    /***
+     * Confirm the anti-concurrency mechanism "expirationCheckInProgressUpdater" works.
+     */
+    @Test
+    void testConcurrentlyExpireMessages() throws Exception {
+        final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        final String cursorName = "s1";
+        admin.topics().createNonPartitionedTopic(topicName);
+        admin.topics().createSubscriptionAsync(topicName, cursorName, MessageId.earliest);
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+        producer.send("1");
+        producer.send("2");
+        producer.send("3");
+        PersistentTopic persistentTopic =
+                (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
+        ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+        ManagedCursorImpl cursor = (ManagedCursorImpl) ml.getCursors().get(cursorName);
+        ManagedCursorImpl spyCursor = Mockito.spy(cursor);
+
+        // Make the mark-deleting delay.
+        CountDownLatch firstFindingCompleted = new CountDownLatch(1);
+        AtomicInteger calledFindPositionCount = new AtomicInteger();
+        doAnswer(invocationOnMock -> {
+            firstFindingCompleted.countDown();
+            ml.getExecutor().execute(() -> {
+                try {
+                    Thread.sleep(3000);
+                    invocationOnMock.callRealMethod();
+                } catch (Throwable ex) {
+                    log.error("Unexpected exception when calling mark delete", ex);
+                }
+            });
+            return true;
+        }).when(spyCursor).asyncMarkDelete(any(Position.class), any(Map.class),
+                any(AsyncCallbacks.MarkDeleteCallback.class), any());
+        doAnswer(invocationOnMock -> {
+            calledFindPositionCount.incrementAndGet();
+            return invocationOnMock.callRealMethod();
+        }).when(spyCursor).asyncFindNewestMatching(any(), any(), any(), any(), any(), any(), anyBoolean());
+        doAnswer(invocationOnMock -> {
+            calledFindPositionCount.incrementAndGet();
+            return invocationOnMock.callRealMethod();
+        }).when(spyCursor).asyncFindNewestMatching(any(), any(), any(), any(), anyBoolean());
+        doAnswer(invocationOnMock -> {
+            calledFindPositionCount.incrementAndGet();
+            return invocationOnMock.callRealMethod();
+        }).when(spyCursor).asyncFindNewestMatching(any(), any(), any(), any());
+
+        // Sleep 2s to make "find(1s)" get a position.
+        Thread.sleep(2000);
+
+        // Start two expire tasks concurrently.
+        PersistentMessageExpiryMonitor
+                monitor = new PersistentMessageExpiryMonitor(persistentTopic, cursorName, spyCursor, null);
+        CompletableFuture<Boolean> expireTask1 = new CompletableFuture<>();
+        new Thread(() -> {
+            expireTask1.complete(monitor.expireMessages(1));
+        }).start();
+        CompletableFuture<Boolean> expireTask2 = new CompletableFuture<>();
+        new Thread(() -> {
+            expireTask2.complete(monitor.expireMessages(1));
+        }).start();
+        firstFindingCompleted.await();
+        CompletableFuture<Boolean> expireTask3 = new CompletableFuture<>();
+        new Thread(() -> {
+            expireTask3.complete(monitor.expireMessages(1));
+        }).start();
+
+        Awaitility.await().untilAsserted(() -> {
+            Assert.assertEquals(monitor.getTotalMessageExpired(), 3);
+        });
+        // Verify: since the other 2 tasks have been prevented, the count of calling find position is 1.
+        Thread.sleep(1000);
+        assertEquals(1, calledFindPositionCount.get());
+
+        // cleanup.
+        producer.close();
+        admin.topics().delete(topicName);
+    }
+}