You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/06/01 18:22:09 UTC
(pulsar) 03/05: [fix][broker] Fix PersistentMessageExpiryMonitor findEntryComplete() method may lose mark-delete properties in race condition (#25803)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c13b393a109726d0439e9e8f933b27d40e7d85a2
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Mon May 25 23:30:50 2026 +0800
[fix][broker] Fix PersistentMessageExpiryMonitor findEntryComplete() method may lose mark-delete properties in race condition (#25803)
(cherry picked from commit 47eec875a81d11a030a3781cbb123e273ced5e88)
---
.../persistent/PersistentMessageExpiryMonitor.java | 3 +-
.../service/PersistentMessageFinderTest.java | 58 ++++++++++++++++++++++
2 files changed, 59 insertions(+), 2 deletions(-)
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 0ad8e8cfe78..05638478c4f 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
@@ -240,8 +240,7 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback, Messag
}
log.info("[{}][{}] Expiring all messages until position {}", topicName, subName, position);
Position prevMarkDeletePos = cursor.getMarkDeletedPosition();
- cursor.asyncMarkDelete(position, cursor.getProperties(), markDeleteCallback,
- cursor.getNumberOfEntriesInBacklog(false));
+ cursor.asyncMarkDelete(position, null, markDeleteCallback, cursor.getNumberOfEntriesInBacklog(false));
if (!Objects.equals(cursor.getMarkDeletedPosition(), prevMarkDeletePos) && subscription != null) {
subscription.updateLastMarkDeleteAdvancedTimestamp();
}
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 d8dc64d0490..d6591d2e0fc 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,6 +19,7 @@
package org.apache.pulsar.broker.service;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -34,11 +35,14 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.UnpooledByteBufAllocator;
import java.lang.reflect.Field;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -1099,4 +1103,58 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertNull(range.getRight());
assertEquals(range.getLeft(), PositionFactory.create(1, 9));
}
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testExpireMessagesNeverLoseMarkDeleteProperties() throws Exception {
+ final String ledgerAndCursorName = "testExpireMessagesNeverLoseMarkDeleteProperties";
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setRetentionSizeInMB(10);
+ config.setRetentionTime(1, TimeUnit.HOURS);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerAndCursorName, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);
+ ManagedCursorImpl spyCursor = spy(cursor);
+
+ Position pos1 = ledger.addEntry(createMessageWrittenToLedger("msg-1"));
+ Position pos2 = ledger.addEntry(createMessageWrittenToLedger("msg-2"));
+
+ CountDownLatch expiryMarkDeleteEnteredLatch = new CountDownLatch(1);
+ CountDownLatch cursorMarkDeleteCompletedLatch = new CountDownLatch(1);
+ CountDownLatch expiryMarkDeleteCompletedLatch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ Map<String, Long> invocationProperties = invocation.getArgument(1);
+ // Pause the expiry-triggered mark-delete so the user markDelete() can complete first.
+ if (invocationProperties == null || invocationProperties.isEmpty()) {
+ expiryMarkDeleteEnteredLatch.countDown();
+ assertTrue(cursorMarkDeleteCompletedLatch.await(5, TimeUnit.SECONDS));
+ try {
+ return invocation.callRealMethod();
+ } finally {
+ expiryMarkDeleteCompletedLatch.countDown();
+ }
+ }
+
+ return invocation.callRealMethod();
+ }).when(spyCursor)
+ .asyncMarkDelete(any(Position.class), nullable(Map.class), any(AsyncCallbacks.MarkDeleteCallback.class),
+ nullable(Object.class));
+
+ PersistentTopic topic = mockPersistentTopic("topicname");
+ PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(topic,
+ spyCursor.getName(), spyCursor, null);
+
+ CompletableFuture.runAsync(() -> monitor.findEntryComplete(pos2, null));
+ assertTrue(expiryMarkDeleteEnteredLatch.await(5, TimeUnit.SECONDS));
+
+ Map<String, Long> properties = new HashMap<>();
+ properties.put("test-property", 1L);
+ spyCursor.markDelete(pos1, properties);
+ cursorMarkDeleteCompletedLatch.countDown();
+
+ assertTrue(expiryMarkDeleteCompletedLatch.await(5, TimeUnit.SECONDS));
+ assertEquals(spyCursor.getMarkDeletedPosition(), pos2);
+ assertEquals(spyCursor.getProperties(), properties);
+ }
}