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/10/16 22:31:40 UTC
(pulsar) 17/24: [fix][broker] Prevent duplicate messages when reconnect races with purge (#26907)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c1de3e7d0ce73633383d68f24e3868091c190668
Author: sinan liu <li...@gmail.com>
AuthorDate: Fri Oct 16 06:56:35 2026 +0800
[fix][broker] Prevent duplicate messages when reconnect races with purge (#26907)
(cherry picked from commit ab0b6d0e648f63989c33680bdc4756a13ad67fd7)
---
.../service/persistent/MessageDeduplication.java | 25 +++--
.../service/persistent/MessageDuplicationTest.java | 108 +++++++++++++++++++++
2 files changed, 125 insertions(+), 8 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 974f6a8f0a04..0c594884b572 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
@@ -30,6 +30,7 @@ import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback;
@@ -126,7 +127,7 @@ public class MessageDeduplication {
private final int maxNumberOfProducers;
// Map used to track the inactive producer along with the timestamp of their last activity
- private final Map<String, Long> inactiveProducers = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, Long> inactiveProducers = new ConcurrentHashMap<>();
private final String replicatorPrefix;
@@ -635,21 +636,29 @@ public class MessageDeduplication {
}
Iterator<Map.Entry<String, Long>> mapIterator = inactiveProducers.entrySet().iterator();
- boolean hasInactive = false;
+ AtomicBoolean hasInactive = new AtomicBoolean();
while (mapIterator.hasNext()) {
- java.util.Map.Entry<String, Long> entry = mapIterator.next();
+ Map.Entry<String, Long> entry = mapIterator.next();
String producerName = entry.getKey();
long lastActiveTimestamp = entry.getValue();
if (lastActiveTimestamp < minimumActiveTimestamp) {
log.info("[{}] Purging dedup information for producer {}", topic.getName(), producerName);
- mapIterator.remove();
- highestSequencedPushed.remove(producerName);
- highestSequencedPersisted.remove(producerName);
- hasInactive = true;
+ // The iterator can retain an expired entry after a reconnect removes it. Recheck inactivity
+ // and remove the sequence IDs atomically with producerAdded/producerRemoved for this name,
+ // so a newly registered producer cannot publish before its old state has been purged.
+ inactiveProducers.computeIfPresent(producerName, (name, inactiveSince) -> {
+ if (inactiveSince < minimumActiveTimestamp) {
+ highestSequencedPushed.remove(name);
+ highestSequencedPersisted.remove(name);
+ hasInactive.set(true);
+ return null;
+ }
+ return inactiveSince;
+ });
}
}
- if (hasInactive && isEnabled()) {
+ if (hasInactive.get() && isEnabled()) {
takeSnapshot(getManagedCursor().getMarkDeletedPosition());
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
index 047a09fcffde..aa0c55abd5db 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
@@ -37,16 +37,27 @@ import static org.testng.Assert.assertTrue;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
import java.lang.reflect.Field;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.layout.PatternLayout;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.qos.AsyncTokenBucket;
@@ -56,8 +67,11 @@ import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.protocol.Commands;
@@ -522,6 +536,100 @@ public class MessageDuplicationTest extends BrokerTestBase {
assertTrue(messageDeduplication.getInactiveProducers().isEmpty());
}
+ @Test(dataProvider = "trueFalse", timeOut = 60000)
+ public void testReconnectDuringInactiveProducerPurge(boolean disconnectAfterReconnect) throws Exception {
+ String topicName = "persistent://prop/ns-abc/" + UUID.randomUUID();
+ String producerName = "reconnecting-producer";
+ try (Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topicName)
+ .startMessageId(MessageId.earliest).create()) {
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName)
+ .producerName(producerName).enableBatching(false).create()) {
+ producer.newMessage().sequenceId(5).value("before-reconnect").send();
+ }
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName)
+ .get().orElseThrow();
+ MessageDeduplication deduplication = topic.getMessageDeduplication();
+ Awaitility.await().until(() -> {
+ Long inactiveSince = deduplication.getInactiveProducers().get(producerName);
+ return inactiveSince != null && inactiveSince < System.currentTimeMillis();
+ });
+ long originalInactiveSince = deduplication.getInactiveProducers().get(producerName);
+
+ CountDownLatch candidateSelected = new CountDownLatch(1);
+ CountDownLatch resumePurge = new CountDownLatch(1);
+ // Pause the actual purge after it reads the expired entry, without replacing its maps or callbacks.
+ AbstractAppender appender = new AbstractAppender("pause-dedup-purge", null,
+ PatternLayout.createDefaultLayout(), false, null) {
+ @Override
+ public void append(LogEvent event) {
+ if (event.getLoggerName().equals(MessageDeduplication.class.getName())
+ && event.getMessage().getFormattedMessage()
+ .equals("[" + topicName + "] Purging dedup information for producer "
+ + producerName)) {
+ candidateSelected.countDown();
+ try {
+ assertTrue(resumePurge.await(30, TimeUnit.SECONDS), "Purge was not resumed");
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ }
+ }
+ };
+ Logger logger = (Logger) LogManager.getLogger(MessageDeduplication.class);
+ ExecutorService executor = Executors.newSingleThreadExecutor(new DefaultThreadFactory("dedup-purge-test"));
+ int inactivityTimeout = conf.getBrokerDeduplicationProducerInactivityTimeoutMinutes();
+ appender.start();
+ logger.addAppender(appender);
+ conf.setBrokerDeduplicationProducerInactivityTimeoutMinutes(0);
+ try {
+ Future<?> purge = executor.submit(topic::checkMessageDeduplicationInfo);
+ assertTrue(candidateSelected.await(30, TimeUnit.SECONDS), "Purge did not select the expired producer");
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName)
+ .producerName(producerName).enableBatching(false).create()) {
+ producer.newMessage().sequenceId(6).value("after-reconnect").send();
+ assertFalse(deduplication.getInactiveProducers().containsKey(producerName));
+ Long refreshedInactiveSince = null;
+ if (disconnectAfterReconnect) {
+ producer.close();
+ Awaitility.await().untilAsserted(() -> {
+ Long inactiveSince = deduplication.getInactiveProducers().get(producerName);
+ assertNotNull(inactiveSince);
+ assertTrue(inactiveSince > originalInactiveSince,
+ "The replacement producer must have a refreshed inactivity timestamp");
+ });
+ refreshedInactiveSince = deduplication.getInactiveProducers().get(producerName);
+ }
+ resumePurge.countDown();
+ purge.get(30, TimeUnit.SECONDS);
+
+ if (disconnectAfterReconnect) {
+ assertEquals(deduplication.getInactiveProducers().get(producerName), refreshedInactiveSince,
+ "The stale purge must preserve the refreshed inactivity record");
+ try (Producer<String> retryProducer = pulsarClient.newProducer(Schema.STRING).topic(topicName)
+ .producerName(producerName).enableBatching(false).create()) {
+ assertEquals(retryProducer.newMessage().sequenceId(6).value("retry").send(),
+ new MessageIdImpl(-1, -1, -1),
+ "A retry after reconnect and disconnect must still be deduplicated");
+ }
+ } else {
+ assertEquals(producer.newMessage().sequenceId(6).value("retry").send(),
+ new MessageIdImpl(-1, -1, -1), "A retry after reconnect must still be deduplicated");
+ }
+ assertEquals(reader.readNext(10, TimeUnit.SECONDS).getValue(), "before-reconnect");
+ assertEquals(reader.readNext(10, TimeUnit.SECONDS).getValue(), "after-reconnect");
+ assertFalse(reader.hasMessageAvailable(), "The retry must not add another stored message");
+ }
+ } finally {
+ resumePurge.countDown();
+ executor.shutdownNow();
+ conf.setBrokerDeduplicationProducerInactivityTimeoutMinutes(inactivityTimeout);
+ logger.removeAppender(appender);
+ appender.stop();
+ assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+ }
@Test
public void testMessageDeduplicationShouldNotWorkForSystemTopic() throws PulsarAdminException {