From 8beac8b12ef7c0ef54529fbb7e4e76c54dea6283 Mon Sep 17 00:00:00 2001 From: lifepuzzlefun Date: Thu, 21 Dec 2023 18:24:08 +0800 Subject: [PATCH] [improve][broker] Avoid record inactiveproducers when deduplication is disable. (#21193) Co-authored-by: Jiwe Guo --- .../persistent/MessageDeduplication.java | 21 ++++++++ .../persistent/MessageDuplicationTest.java | 49 ++++++++++++++++++- 2 files changed, 69 insertions(+), 1 deletion(-) 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 238dc740509b1..802dd91796127 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 @@ -474,6 +474,10 @@ private boolean isDeduplicationEnabled() { * Topic will call this method whenever a producer connects. */ public void producerAdded(String producerName) { + if (!isEnabled()) { + return; + } + // Producer is no-longer inactive inactiveProducers.remove(producerName); } @@ -482,6 +486,10 @@ public void producerAdded(String producerName) { * Topic will call this method whenever a producer disconnects. */ public void producerRemoved(String producerName) { + if (!isEnabled()) { + return; + } + // Producer is no-longer active inactiveProducers.put(producerName, System.currentTimeMillis()); } @@ -493,6 +501,14 @@ public synchronized void purgeInactiveProducers() { long minimumActiveTimestamp = System.currentTimeMillis() - TimeUnit.MINUTES .toMillis(pulsar.getConfiguration().getBrokerDeduplicationProducerInactivityTimeoutMinutes()); + // if not enabled just clear all inactive producer record. + if (!isEnabled()) { + if (!inactiveProducers.isEmpty()) { + inactiveProducers.clear(); + } + return; + } + Iterator> mapIterator = inactiveProducers.entrySet().iterator(); boolean hasInactive = false; while (mapIterator.hasNext()) { @@ -545,5 +561,10 @@ ManagedCursor getManagedCursor() { return managedCursor; } + @VisibleForTesting + Map getInactiveProducers() { + return inactiveProducers; + } + private static final Logger log = LoggerFactory.getLogger(MessageDeduplication.class); } 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 81a565f9d6feb..402b5c4972ce2 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 @@ -32,6 +32,7 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import io.netty.buffer.ByteBuf; import io.netty.channel.EventLoopGroup; import java.lang.reflect.Field; @@ -47,17 +48,24 @@ import org.apache.pulsar.broker.resources.PulsarResources; import org.apache.pulsar.broker.service.BacklogQuotaManager; 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.api.Producer; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.broker.qos.AsyncTokenBucket; import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap; import org.apache.pulsar.compaction.CompactionServiceFactory; +import org.awaitility.Awaitility; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @Slf4j @Test(groups = "broker") -public class MessageDuplicationTest { +public class MessageDuplicationTest extends BrokerTestBase { private static final int BROKER_DEDUPLICATION_ENTRIES_INTERVAL = 10; private static final int BROKER_DEDUPLICATION_MAX_NUMBER_PRODUCERS = 10; @@ -443,4 +451,43 @@ public void completed(Exception e, long ledgerId, long entryId) { } }); } + + @BeforeMethod(alwaysRun = true) + @Override + protected void setup() throws Exception { + this.conf.setBrokerDeduplicationEnabled(true); + super.baseSetup(); + } + + @AfterMethod(alwaysRun = true) + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + @Test + public void testMessageDeduplication() throws Exception { + String topicName = "persistent://prop/ns-abc/testMessageDeduplication"; + String producerName = "test-producer"; + Producer producer = pulsarClient + .newProducer(Schema.STRING) + .producerName(producerName) + .topic(topicName) + .create(); + final PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopicIfExists(topicName).get().orElse(null); + assertNotNull(persistentTopic); + final MessageDeduplication messageDeduplication = persistentTopic.getMessageDeduplication(); + assertFalse(messageDeduplication.getInactiveProducers().containsKey(producerName)); + producer.close(); + Awaitility.await().untilAsserted(() -> assertTrue(messageDeduplication.getInactiveProducers().containsKey(producerName))); + admin.topicPolicies().setDeduplicationStatus(topicName, false); + Awaitility.await().untilAsserted(() -> { + final Boolean deduplicationStatus = admin.topicPolicies().getDeduplicationStatus(topicName); + Assert.assertNotNull(deduplicationStatus); + Assert.assertFalse(deduplicationStatus); + }); + messageDeduplication.purgeInactiveProducers(); + assertTrue(messageDeduplication.getInactiveProducers().isEmpty()); + } }