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/20 15:52:51 UTC
(pulsar) 05/12: [improve][broker] Record the deduplication's persisted sequence IDs in place (#26822)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 597f1bde968475f7a9e13d0e537437fcb035d6fb
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Oct 20 15:34:31 2026 +0300
[improve][broker] Record the deduplication's persisted sequence IDs in place (#26822)
(cherry picked from commit b038f30042c847af7f85cf3271bac5f1fbaabcd4)
---
.../service/persistent/MessageDeduplication.java | 92 ++++++++++++---
.../service/persistent/MessageDuplicationTest.java | 129 +++++++++++++++++++--
.../service/persistent/TopicDuplicationTest.java | 10 +-
3 files changed, 197 insertions(+), 34 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 9bb2eb5d52d8..386f4aca4791 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
@@ -23,12 +23,12 @@ import static org.apache.bookkeeper.mledger.util.ManagedLedgerUtils.openCursor;
import static org.apache.pulsar.client.impl.GeoReplicationProducerImpl.MSG_PROP_IS_REPL_MARKER;
import static org.apache.pulsar.client.impl.GeoReplicationProducerImpl.MSG_PROP_REPL_SOURCE_POSITION;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
import io.github.merlimat.slog.Logger;
import io.netty.buffer.ByteBuf;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -115,7 +115,21 @@ public class MessageDeduplication {
// Map that contains the highest sequenceId that have been persistent by each producers. The map will be updated
// after the messages are persisted
@VisibleForTesting
- final ConcurrentMap<String, Long> highestSequencedPersisted = new ConcurrentHashMap<>();
+ final ConcurrentMap<String, PersistedSequenceId> highestSequencedPersisted = new ConcurrentHashMap<>();
+
+ /**
+ * The highest sequence ID persisted for a producer. It's recorded for every persisted message, normally on the
+ * topic's managed-ledger thread, so the value is updated in place: replacing the map's entry would lock a bin of
+ * the map and box the value for every message. See {@link #setHighestSequencedPersisted(String, long)} for how an
+ * update orders with a removal of the entry.
+ */
+ static final class PersistedSequenceId {
+ volatile long value;
+
+ PersistedSequenceId(long value) {
+ this.value = value;
+ }
+ }
// Number of persisted entries after which to store a snapshot of the sequence ids map
private final int snapshotInterval;
@@ -271,7 +285,7 @@ public class MessageDeduplication {
managedCursor.getProperties().forEach((k, v) -> {
producerRemoved(k);
highestSequencedPushed.put(k, v);
- highestSequencedPersisted.put(k, v);
+ setHighestSequencedPersisted(k, v);
});
// Replay all the entries and apply all the sequence ids updates
final long startTimeReplayDedup = System.currentTimeMillis();
@@ -283,7 +297,7 @@ public class MessageDeduplication {
final var producerName = metadata.getProducerName();
final var sequenceId = Math.max(metadata.getHighestSequenceId(), metadata.getSequenceId());
highestSequencedPushed.put(producerName, sequenceId);
- highestSequencedPersisted.put(producerName, sequenceId);
+ setHighestSequencedPersisted(producerName, sequenceId);
producerRemoved(producerName);
}).thenCompose(optPosition -> {
if (optPosition.isEmpty()) {
@@ -433,8 +447,8 @@ public class MessageDeduplication {
// "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
// is a dup or not we should return an error to the producer for the latter case so that it can retry
// at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
+ Long lastSequenceLIdPersisted = getHighestSequencedPersisted(lastSequenceLIdKey);
+ Long lastSequenceEIdPersisted = getHighestSequencedPersisted(lastSequenceEIdKey);
log.debug()
.attr("producerName", publishContext.getProducerName())
.attr("replSequenceLId", replSequenceLId)
@@ -503,7 +517,7 @@ public class MessageDeduplication {
// If current message's seq id is between lastSequenceIdPersisted and
// lastSequenceIdPushed, then we cannot be sure whether the message is a dup or not
// we should return an error to the producer for the latter case so that it can retry at a future time
- Long lastSequenceIdPersisted = highestSequencedPersisted.get(producerName);
+ Long lastSequenceIdPersisted = getHighestSequencedPersisted(producerName);
if (lastSequenceIdPersisted != null && sequenceId <= lastSequenceIdPersisted) {
return MessageDupStatus.Dup;
} else {
@@ -564,8 +578,8 @@ public class MessageDeduplication {
long replSequenceEId = positionPair[1];
String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- highestSequencedPersisted.put(lastSequenceLIdKey, replSequenceLId);
- highestSequencedPersisted.put(lastSequenceEIdKey, replSequenceEId);
+ setHighestSequencedPersisted(lastSequenceLIdKey, replSequenceLId);
+ setHighestSequencedPersisted(lastSequenceEIdKey, replSequenceEId);
increaseSnapshotCounterAndTakeSnapshotIfNeeded(position);
}
@@ -581,7 +595,7 @@ public class MessageDeduplication {
}
Boolean isLastChunk = (Boolean) publishContext.getProperty(IS_LAST_CHUNK);
if (isLastChunk == null || isLastChunk) {
- highestSequencedPersisted.put(producerName, Math.max(highestSequenceId, sequenceId));
+ setHighestSequencedPersisted(producerName, Math.max(highestSequenceId, sequenceId));
}
increaseSnapshotCounterAndTakeSnapshotIfNeeded(position);
}
@@ -604,9 +618,8 @@ public class MessageDeduplication {
}
highestSequencedPushed.clear();
- for (String producer : highestSequencedPersisted.keySet()) {
- highestSequencedPushed.put(producer, highestSequencedPersisted.get(producer));
- }
+ highestSequencedPersisted.forEach((producer, persisted) ->
+ highestSequencedPushed.put(producer, persisted.value));
}
private CompletableFuture<Void> takeSnapshot(Position position) {
@@ -619,12 +632,7 @@ public class MessageDeduplication {
return CompletableFuture.completedFuture(null);
}
- Map<String, Long> snapshot = new TreeMap<>();
- highestSequencedPersisted.forEach((producerName, sequenceId) -> {
- if (snapshot.size() < maxNumberOfProducers) {
- snapshot.put(producerName, sequenceId);
- }
- });
+ Map<String, Long> snapshot = snapshotPersistedSequenceIds();
final var cursor = managedCursor;
if (cursor == null) {
@@ -720,6 +728,52 @@ public class MessageDeduplication {
}
}
+ /**
+ * The highest persisted sequence IDs of up to {@code maxNumberOfProducers} producers, to store in the cursor's
+ * properties. The properties don't need an order, so the producers' names aren't sorted.
+ */
+ Map<String, Long> snapshotPersistedSequenceIds() {
+ Map<String, Long> snapshot = Maps.newHashMapWithExpectedSize(
+ Math.max(0, Math.min(highestSequencedPersisted.size(), maxNumberOfProducers)));
+ highestSequencedPersisted.forEach((producerName, persisted) -> {
+ if (snapshot.size() < maxNumberOfProducers) {
+ snapshot.put(producerName, persisted.value);
+ }
+ });
+ return snapshot;
+ }
+
+ /**
+ * Records the highest sequence ID persisted for a producer, updating its entry in place.
+ *
+ * <p>A purge of inactive producers or a clear that removes the entry after the lookup orders after the update:
+ * the update's value is removed with the entry, as a put of the value before the removal would have been. The
+ * entry isn't put back, so a removal that has completed stays in effect, and a producer that reuses the name
+ * doesn't inherit a sequence ID it hasn't persisted.
+ */
+ void setHighestSequencedPersisted(String producerName, long sequenceId) {
+ setHighestSequencedPersisted(highestSequencedPersisted, producerName, sequenceId);
+ }
+
+ @VisibleForTesting
+ static void setHighestSequencedPersisted(ConcurrentMap<String, PersistedSequenceId> highestSequencedPersisted,
+ String producerName, long sequenceId) {
+ PersistedSequenceId persisted = highestSequencedPersisted.get(producerName);
+ if (persisted == null) {
+ highestSequencedPersisted.put(producerName, new PersistedSequenceId(sequenceId));
+ } else {
+ persisted.value = sequenceId;
+ }
+ }
+
+ /**
+ * Returns the highest sequence ID persisted for a producer, or null when there's none.
+ */
+ Long getHighestSequencedPersisted(String producerName) {
+ PersistedSequenceId persisted = highestSequencedPersisted.get(producerName);
+ return persisted != null ? persisted.value : null;
+ }
+
public long getLastPublishedSequenceId(String producerName) {
Long sequenceId = highestSequencedPushed.get(producerName);
return sequenceId != null ? sequenceId : -1;
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 26627a37f86d..43e2e3e5b608 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
@@ -33,7 +33,9 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
+import com.google.common.collect.ForwardingConcurrentMap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.EventLoopGroup;
@@ -45,6 +47,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -159,7 +162,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
assertEquals(lastSequenceIdPushed.longValue(), 5);
// update highest sequence persisted
- messageDeduplication.highestSequencedPersisted.put(producerName1, 0L);
+ messageDeduplication.setHighestSequencedPersisted(producerName1, 0L);
byteBuf1 = getMessage(producerName1, 0);
publishContext1 = getPublishContext(producerName1, 0);
@@ -173,7 +176,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
// update highest sequence persisted
messageDeduplication.highestSequencedPushed.put(producerName1, 0L);
- messageDeduplication.highestSequencedPersisted.put(producerName1, 0L);
+ messageDeduplication.setHighestSequencedPersisted(producerName1, 0L);
byteBuf1 = getMessage(producerName1, 0);
publishContext1 = getPublishContext(producerName1, 1, 5);
status = messageDeduplication.isDuplicate(publishContext1, byteBuf1);
@@ -191,6 +194,112 @@ public class MessageDuplicationTest extends BrokerTestBase {
assertEquals(lastSequenceIdPushed.longValue(), 5);
}
+ @Test
+ public void testSnapshotKeepsTheLatestSequenceIdsOfUpToMaxProducers() {
+ MessageDeduplication messageDeduplication = messageDeduplication(BROKER_DEDUPLICATION_MAX_NUMBER_PRODUCERS);
+ for (int i = 0; i < 15; i++) {
+ messageDeduplication.setHighestSequencedPersisted("producer" + i, i);
+ }
+ MessageDeduplication.PersistedSequenceId persisted =
+ messageDeduplication.highestSequencedPersisted.get("producer0");
+ // updated in place
+ for (int i = 0; i < 15; i++) {
+ messageDeduplication.setHighestSequencedPersisted("producer" + i, 100 + i);
+ }
+ assertSame(messageDeduplication.highestSequencedPersisted.get("producer0"), persisted);
+ Map<String, Long> snapshot = messageDeduplication.snapshotPersistedSequenceIds();
+ assertEquals(snapshot.size(), BROKER_DEDUPLICATION_MAX_NUMBER_PRODUCERS);
+ snapshot.forEach((producerName, sequenceId) ->
+ assertEquals((long) sequenceId, 100 + Long.parseLong(producerName.substring("producer".length()))));
+
+ for (int maxNumberOfProducers : new int[] {0, -1, -10}) {
+ MessageDeduplication capped = messageDeduplication(maxNumberOfProducers);
+ capped.setHighestSequencedPersisted("producer", 1);
+ assertTrue(capped.snapshotPersistedSequenceIds().isEmpty());
+ }
+ }
+
+ @Test
+ public void testAPurgeRacingAnUpdateOrdersAfterIt() {
+ // the purge runs between the update's lookup and its write, or after the update
+ for (boolean purgeDuringUpdate : new boolean[] {true, false}) {
+ PulsarService pulsarService = mock(PulsarService.class);
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ serviceConfiguration.setBrokerDeduplicationEntriesInterval(BROKER_DEDUPLICATION_ENTRIES_INTERVAL);
+ serviceConfiguration.setBrokerDeduplicationMaxNumberOfProducers(BROKER_DEDUPLICATION_MAX_NUMBER_PRODUCERS);
+ serviceConfiguration.setReplicatorPrefix(REPLICATOR_PREFIX);
+ serviceConfiguration.setBrokerDeduplicationProducerInactivityTimeoutMinutes(1);
+ doReturn(serviceConfiguration).when(pulsarService).getConfiguration();
+ MessageDeduplication messageDeduplication = spyWithClassAndConstructorArgs(MessageDeduplication.class,
+ pulsarService, mock(PersistentTopic.class), mock(ManagedLedger.class));
+ doReturn(true).when(messageDeduplication).isEnabled();
+ doReturn(mock(ManagedCursor.class)).when(messageDeduplication).getManagedCursor();
+
+ // the producer has persisted up to 4 and publishes 5, then expires while 5 is being persisted
+ String producerName = "producer";
+ messageDeduplication.setHighestSequencedPersisted(producerName, 4);
+ assertEquals(messageDeduplication.isDuplicate(getPublishContext(producerName, 5), null),
+ MessageDeduplication.MessageDupStatus.NotDup);
+ messageDeduplication.producerRemoved(producerName);
+ messageDeduplication.getInactiveProducers().put(producerName, System.currentTimeMillis() - 120_000);
+ ConcurrentMap<String, MessageDeduplication.PersistedSequenceId> purgedOnLookup =
+ new ForwardingConcurrentMap<>() {
+ @Override
+ protected ConcurrentMap<String, MessageDeduplication.PersistedSequenceId> delegate() {
+ return messageDeduplication.highestSequencedPersisted;
+ }
+
+ @Override
+ public MessageDeduplication.PersistedSequenceId get(Object key) {
+ MessageDeduplication.PersistedSequenceId value = super.get(key);
+ messageDeduplication.purgeInactiveProducers();
+ return value;
+ }
+ };
+ if (purgeDuringUpdate) {
+ MessageDeduplication.setHighestSequencedPersisted(purgedOnLookup, producerName, 5);
+ } else {
+ messageDeduplication.setHighestSequencedPersisted(producerName, 5);
+ messageDeduplication.purgeInactiveProducers();
+ }
+ assertFalse(messageDeduplication.highestSequencedPersisted.containsKey(producerName));
+ assertFalse(messageDeduplication.highestSequencedPushed.containsKey(producerName));
+
+ // a producer with the same name restarts at 0: a retry before 0 is persisted isn't a known duplicate
+ assertEquals(messageDeduplication.isDuplicate(getPublishContext(producerName, 0), null),
+ MessageDeduplication.MessageDupStatus.NotDup);
+ assertEquals(messageDeduplication.isDuplicate(getPublishContext(producerName, 0), null),
+ MessageDeduplication.MessageDupStatus.Unknown);
+ messageDeduplication.setHighestSequencedPersisted(producerName, 0);
+ assertEquals(messageDeduplication.isDuplicate(getPublishContext(producerName, 0), null),
+ MessageDeduplication.MessageDupStatus.Dup);
+ }
+ }
+
+ @Test
+ public void testRecordingARemovedProducerAddsANewEntry() {
+ MessageDeduplication messageDeduplication = messageDeduplication(BROKER_DEDUPLICATION_MAX_NUMBER_PRODUCERS);
+ messageDeduplication.setHighestSequencedPersisted("producer", 1);
+ MessageDeduplication.PersistedSequenceId persisted =
+ messageDeduplication.highestSequencedPersisted.get("producer");
+ // a purge of inactive producers removes the entry
+ messageDeduplication.highestSequencedPersisted.remove("producer");
+ messageDeduplication.setHighestSequencedPersisted("producer", 2);
+ assertEquals(messageDeduplication.highestSequencedPersisted.get("producer").value, 2L);
+ assertEquals(persisted.value, 1L);
+ assertEquals(messageDeduplication.snapshotPersistedSequenceIds(), Map.of("producer", 2L));
+ }
+
+ private static MessageDeduplication messageDeduplication(int maxNumberOfProducers) {
+ PulsarService pulsarService = mock(PulsarService.class);
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ serviceConfiguration.setBrokerDeduplicationEntriesInterval(BROKER_DEDUPLICATION_ENTRIES_INTERVAL);
+ serviceConfiguration.setBrokerDeduplicationMaxNumberOfProducers(maxNumberOfProducers);
+ serviceConfiguration.setReplicatorPrefix(REPLICATOR_PREFIX);
+ doReturn(serviceConfiguration).when(pulsarService).getConfiguration();
+ return new MessageDeduplication(pulsarService, mock(PersistentTopic.class), mock(ManagedLedger.class));
+ }
+
@Test
public void testConcurrentDuplicateCheckForSameProducer() throws Exception {
PulsarService pulsarService = mock(PulsarService.class);
@@ -360,7 +469,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
Long lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 0);
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName1);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 0);
@@ -372,7 +481,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
assertEquals(lastSequenceIdPushed.longValue(), 1);
byteBuf2.release();
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName2);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName2);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 1);
byteBuf1.release();
@@ -385,7 +494,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 1);
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName1);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 1);
byteBuf1.release();
@@ -398,7 +507,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 5);
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName1);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 5);
byteBuf1.release();
@@ -423,7 +532,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 6);
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName1);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 5);
byteBuf1.release();
@@ -452,11 +561,11 @@ public class MessageDuplicationTest extends BrokerTestBase {
assertEquals(messageDeduplication.highestSequencedPersisted.size(), 2);
lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
assertEquals(lastSequenceIdPushed.longValue(), 6);
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName1);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName1);
assertEquals(lastSequenceIdPushed.longValue(), 6);
lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName2);
assertEquals(lastSequenceIdPushed.longValue(), 1);
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName2);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName2);
assertEquals(lastSequenceIdPushed.longValue(), 1);
verify(messageDeduplication, times(1)).resetHighestSequenceIdPushed();
byteBuf1.release();
@@ -481,7 +590,7 @@ public class MessageDuplicationTest extends BrokerTestBase {
lastSequenceIdPushed = messageDeduplication.highestSequencedPushed.get(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 8);
- lastSequenceIdPushed = messageDeduplication.highestSequencedPersisted.get(producerName1);
+ lastSequenceIdPushed = messageDeduplication.getHighestSequencedPersisted(producerName1);
assertNotNull(lastSequenceIdPushed);
assertEquals(lastSequenceIdPushed.longValue(), 8);
byteBuf1.release();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java
index bb9cfb99f89b..42371582f541 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/TopicDuplicationTest.java
@@ -214,7 +214,7 @@ public class TopicDuplicationTest extends ProducerConsumerBase {
assertNotNull(messageDeduplication.highestSequencedPushed);
long seqId = messageDeduplication.getLastPublishedSequenceId(producerName);
assertEquals(seqId, maxSeq);
- assertEquals(messageDeduplication.highestSequencedPersisted.get(producerName).longValue(), maxSeq);
+ assertEquals(messageDeduplication.getHighestSequencedPersisted(producerName).longValue(), maxSeq);
assertEquals(messageDeduplication.highestSequencedPushed.get(producerName).longValue(), maxSeq);
}).get();
}
@@ -266,7 +266,7 @@ public class TopicDuplicationTest extends ProducerConsumerBase {
countDownLatch.await();
PersistentTopic persistentTopic =
(PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
- long seqId = persistentTopic.getMessageDeduplication().highestSequencedPersisted.get(producerName);
+ long seqId = persistentTopic.getMessageDeduplication().getHighestSequencedPersisted(producerName);
Position position = persistentTopic.getMessageDeduplication().getManagedCursor()
.getManagedLedger().getLastConfirmedEntry();
assertEquals(seqId, msgNum - 1);
@@ -375,7 +375,7 @@ public class TopicDuplicationTest extends ProducerConsumerBase {
countDownLatch.await();
PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
.getTopicIfExists(topicName).get().get();
- long seqId = persistentTopic.getMessageDeduplication().highestSequencedPersisted.get(producerName);
+ long seqId = persistentTopic.getMessageDeduplication().getHighestSequencedPersisted(producerName);
Position position = persistentTopic.getMessageDeduplication().getManagedCursor().getManagedLedger()
.getLastConfirmedEntry();
assertEquals(seqId, msgNum - 1);
@@ -457,7 +457,7 @@ public class TopicDuplicationTest extends ProducerConsumerBase {
countDownLatch.await();
PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
.getTopicIfExists(topicName).get().get();
- long seqId = persistentTopic.getMessageDeduplication().highestSequencedPersisted.get(producerName);
+ long seqId = persistentTopic.getMessageDeduplication().getHighestSequencedPersisted(producerName);
Position position = persistentTopic.getMessageDeduplication().getManagedCursor()
.getManagedLedger().getLastConfirmedEntry();
assertEquals(seqId, msgNum - 1);
@@ -514,7 +514,7 @@ public class TopicDuplicationTest extends ProducerConsumerBase {
ManagedCursor managedCursor = persistentTopic.getMessageDeduplication().getManagedCursor();
Position markDeletedPosition = managedCursor.getMarkDeletedPosition();
- long seqId = persistentTopic.getMessageDeduplication().highestSequencedPersisted.get(producerName);
+ long seqId = persistentTopic.getMessageDeduplication().getHighestSequencedPersisted(producerName);
Position position = persistentTopic.getMessageDeduplication().getManagedCursor()
.getManagedLedger().getLastConfirmedEntry();
assertEquals(seqId, msgNum - 1);