You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/17 00:07:10 UTC
(pulsar) branch master updated: [fix][broker] Apply dedup disable after recovery (#25982)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 31bb46606d67 [fix][broker] Apply dedup disable after recovery (#25982)
31bb46606d67 is described below
commit 31bb46606d6729ed5c1216639db3174b28532144
Author: sinan liu <li...@gmail.com>
AuthorDate: Sat Oct 17 08:07:03 2026 +0800
[fix][broker] Apply dedup disable after recovery (#25982)
---
.../service/persistent/MessageDeduplication.java | 5 +-
.../broker/BrokerMessageDeduplicationTest.java | 84 +++++++++++++++++++++-
2 files changed, 85 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
index a9ced74e432f..9bb2eb5d52d8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
@@ -162,6 +162,9 @@ public class MessageDeduplication {
boolean shouldBeEnabled = topic.isDeduplicationEnabled();
synchronized (this) {
if (status == Status.Recovering) {
+ if (!shouldBeEnabled) {
+ return statusChangeFuture.handle((__, e) -> null).thenCompose(__ -> checkStatus());
+ }
return statusChangeFuture;
}
if (status == Status.Removing) {
@@ -192,7 +195,7 @@ public class MessageDeduplication {
}, null);
}
- if (status == Status.Enabled && !shouldBeEnabled) {
+ if ((status == Status.Enabled || status == Status.Failed) && !shouldBeEnabled) {
// Disabled deduping
CompletableFuture<Void> future = new CompletableFuture<>();
status = Status.Removing;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
index f36caf87e6c7..bc8af54570d1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BrokerMessageDeduplicationTest.java
@@ -56,6 +56,7 @@ import org.testng.annotations.Test;
public class BrokerMessageDeduplicationTest {
private ManagedLedger managedLedger;
+ private PersistentTopic topic;
private MessageDeduplication deduplication;
private ScheduledExecutorService executor;
@@ -68,9 +69,9 @@ public class BrokerMessageDeduplicationTest {
executor = Executors.newScheduledThreadPool(1, new ExecutorProvider.ExtendedThreadFactory("pulsar"));
doReturn(executor).when(pulsarService).getExecutor();
managedLedger = mock(ManagedLedger.class);
- final var mockTopic = mock(PersistentTopic.class);
- doReturn(true).when(mockTopic).isDeduplicationEnabled();
- deduplication = spy(new MessageDeduplication(pulsarService, mockTopic, managedLedger));
+ topic = mock(PersistentTopic.class);
+ doReturn(true).when(topic).isDeduplicationEnabled();
+ deduplication = spy(new MessageDeduplication(pulsarService, topic, managedLedger));
doReturn(true).when(deduplication).isEnabled();
}
@@ -163,6 +164,83 @@ public class BrokerMessageDeduplicationTest {
verify(managedLedger, times(1)).asyncOpenCursor(any(), any(), any());
}
+ @Test
+ public void checkStatusDisablesAfterRecoveryCompletesWhenPolicyChanges() throws Exception {
+ final var cursor = mock(ManagedCursor.class);
+ final var shouldEnableDeduplication = new AtomicReference<>(true);
+ final var openCursorCallback = new AtomicReference<AsyncCallbacks.OpenCursorCallback>();
+ doAnswer(invocation -> shouldEnableDeduplication.get()).when(topic).isDeduplicationEnabled();
+ doAnswer(invocation -> {
+ openCursorCallback.set(invocation.getArgument(1));
+ return null;
+ }).when(managedLedger).asyncOpenCursor(any(), any(), any());
+ doAnswer(invocation -> {
+ ((AsyncCallbacks.DeleteCursorCallback) invocation.getArgument(1)).deleteCursorComplete(null);
+ return null;
+ }).when(managedLedger).asyncDeleteCursor(any(), any(), any());
+ doReturn(Map.of("from-snapshot", 10L)).when(cursor).getProperties();
+ doReturn(false).when(cursor).hasMoreEntries();
+
+ final var enableFuture = deduplication.checkStatus();
+ assertFalse(enableFuture.isDone());
+ assertEquals(String.valueOf(deduplication.getStatus()), "Recovering");
+
+ shouldEnableDeduplication.set(false);
+ final var disableFuture = deduplication.checkStatus();
+ assertFalse(disableFuture.isDone());
+ verify(managedLedger, times(1)).asyncOpenCursor(any(), any(), any());
+
+ openCursorCallback.get().openCursorComplete(cursor, null);
+ enableFuture.get(3, TimeUnit.SECONDS);
+ disableFuture.get(3, TimeUnit.SECONDS);
+
+ assertEquals(String.valueOf(deduplication.getStatus()), "Disabled");
+ assertEquals(deduplication.getLastPublishedSequenceId("from-snapshot"), -1L);
+ verify(managedLedger, times(1)).asyncDeleteCursor(any(), any(), any());
+ }
+
+ @Test
+ public void checkStatusDisablesAfterRecoveryFailsWhenPolicyChanges() throws Exception {
+ final var cursor = mock(ManagedCursor.class);
+ final var shouldEnableDeduplication = new AtomicReference<>(true);
+ final var openCursorCallback = new AtomicReference<AsyncCallbacks.OpenCursorCallback>();
+ doAnswer(invocation -> shouldEnableDeduplication.get()).when(topic).isDeduplicationEnabled();
+ doAnswer(invocation -> {
+ openCursorCallback.set(invocation.getArgument(1));
+ return null;
+ }).when(managedLedger).asyncOpenCursor(any(), any(), any());
+ doAnswer(invocation -> {
+ ((AsyncCallbacks.DeleteCursorCallback) invocation.getArgument(1)).deleteCursorComplete(null);
+ return null;
+ }).when(managedLedger).asyncDeleteCursor(any(), any(), any());
+ doReturn(Map.of()).when(cursor).getProperties();
+ doReturn(true).when(cursor).hasMoreEntries();
+ doAnswer(invocation -> {
+ throw new RuntimeException("asyncReadEntries failed");
+ }).when(cursor).asyncReadEntries(anyInt(), anyLong(), any(), any(), any());
+
+ final var enableFuture = deduplication.checkStatus();
+ assertFalse(enableFuture.isDone());
+ assertEquals(String.valueOf(deduplication.getStatus()), "Recovering");
+
+ shouldEnableDeduplication.set(false);
+ final var disableFuture = deduplication.checkStatus();
+ assertFalse(disableFuture.isDone());
+
+ openCursorCallback.get().openCursorComplete(cursor, null);
+ try {
+ enableFuture.get(3, TimeUnit.SECONDS);
+ fail();
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof RuntimeException);
+ assertTrue(e.getMessage().contains("asyncReadEntries failed"));
+ }
+ disableFuture.get(3, TimeUnit.SECONDS);
+
+ assertEquals(String.valueOf(deduplication.getStatus()), "Disabled");
+ verify(managedLedger, times(1)).asyncDeleteCursor(any(), any(), any());
+ }
+
@Test
public void checkStatusRetriesAfterFailedEnable() throws Exception {
final var cursor = mock(ManagedCursor.class);