diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java index 643b7cad9a645..f5199c14b2b57 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java +++ b/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 diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java index bbffe08395233..5f30389fb1115 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java +++ b/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(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitorTest.java new file mode 100644 index 0000000000000..5535561a5fa83 --- /dev/null +++ b/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 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 expireTask1 = new CompletableFuture<>(); + new Thread(() -> { + expireTask1.complete(monitor.expireMessages(1)); + }).start(); + CompletableFuture expireTask2 = new CompletableFuture<>(); + new Thread(() -> { + expireTask2.complete(monitor.expireMessages(1)); + }).start(); + firstFindingCompleted.await(); + CompletableFuture 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); + } +}