From 57c0a46aa5c509db4eecad5b8afde78c8f92cef7 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 24 Jul 2024 17:12:38 +0800 Subject: [PATCH 01/17] addProducerDuplicationWithReceiptLost test. --- .../client/api/DeduplicationEndToEndTest.java | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java new file mode 100644 index 0000000000000..a05c892a6dc79 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -0,0 +1,109 @@ +package org.apache.pulsar.client.api; + +import org.apache.pulsar.broker.service.PulsarCommandSender; +import org.apache.pulsar.broker.service.ServerCnx; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.mockito.Mockito; +import org.testng.Assert; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import java.lang.reflect.Field; +import java.util.concurrent.TimeUnit; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; + +/** + * This test class is used to point out cases where message duplication can occur, + * producer idempotency features can be used to solve which cases and can't solve which cases. + */ +public class DeduplicationEndToEndTest extends ProducerConsumerBase { + + @BeforeClass + @Override + protected void setup() throws Exception { + super.internalSetup(); + super.producerBaseSetup(); + } + + @AfterClass(alwaysRun = true) + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + /** + * simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. User receives the exception and sends the same message again. + * The message is duplicated in the topic. + * @throws Exception + */ + @Test + public void testProducerDuplicationWithReceiptLost() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test"; + admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); + + // Create producer with deduplication enabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic, false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // use reflection to spy the commandSender + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); + commandSenderField.set(serverCnx, spyCommandSender); + + // disable the send receipt + Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), + Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + + // send a message + producer.sendAsync("test".getBytes()).thenRun(() -> { + // should not enter here + Assert.fail(); + }).exceptionally(e -> { + // do not receive the ack, should enter here + return null; + }).get(); + + + // set back the send receipt + commandSenderField.set(serverCnx, commandSender); + + // user receive the exception, send the same message again + producer.sendAsync("test".getBytes()).exceptionally(e -> { + // should not enter here + Assert.fail(); + return null; + }).get(); + + // consume the message, there are two messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + + // clean up + producer.close(); + consumer.close(); + } + + + + + +} From cf9851d555d939816951127056269941c13143f2 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 24 Jul 2024 18:18:22 +0800 Subject: [PATCH 02/17] add test. --- .../client/api/DeduplicationEndToEndTest.java | 147 ++++++++++++++++++ 1 file changed, 147 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index a05c892a6dc79..0cbae87f878d0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -43,6 +43,76 @@ protected void cleanup() throws Exception { @Test public void testProducerDuplicationWithReceiptLost() throws Exception { final String topic = "persistent://my-property/my-ns/deduplication-test"; + admin.namespaces().setDeduplicationStatus("my-property/my-ns", false); + + // Create producer with deduplication enabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic, false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // use reflection to spy the commandSender + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); + commandSenderField.set(serverCnx, spyCommandSender); + + // disable the send receipt + Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), + Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + + // send a message + producer.sendAsync("test".getBytes()).thenRun(() -> { + // should not enter here + Assert.fail(); + }).exceptionally(e -> { + // do not receive the ack, should enter here + return null; + }).get(); + + + // set back the send receipt + commandSenderField.set(serverCnx, commandSender); + + // user receive the exception, send the same message again + producer.sendAsync("test".getBytes()).exceptionally(e -> { + // should not enter here + Assert.fail(); + return null; + }).get(); + + // consume the message, there are two messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + + // clean up + producer.close(); + consumer.close(); + } + + + /** + * simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. User receives the exception and sends the same message again. + * With deduplication enabled, the message is duplicated too! Because the message newly sent has a + * different sequence id. + * @throws Exception + */ + @Test + public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled"; admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); // Create producer with deduplication enabled @@ -83,6 +153,7 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { commandSenderField.set(serverCnx, commandSender); // user receive the exception, send the same message again + // though the message content is the same, the sequence id is different, so the message is duplicated producer.sendAsync("test".getBytes()).exceptionally(e -> { // should not enter here Assert.fail(); @@ -93,9 +164,11 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { Message message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), 0); message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), 1); // clean up producer.close(); @@ -103,6 +176,80 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { } + /** + * simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. User receives the exception and sends the same message again. + * With deduplication enabled, the message is duplicated too! Because the message newly sent has a + * different sequence id. + * @throws Exception + */ + @Test + public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled2"; + admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); + + // Create producer with deduplication enabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic, false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // use reflection to spy the commandSender + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); + commandSenderField.set(serverCnx, spyCommandSender); + + // disable the send receipt + Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), + Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + + // send a message + TypedMessageBuilder typeMessages = producer.newMessage().value("test".getBytes()); + typeMessages.sendAsync().thenRun(() -> { + // should not enter here + Assert.fail(); + }).exceptionally(e -> { + // do not receive the ack, should enter here + return null; + }).get(); + + + // set back the send receipt + commandSenderField.set(serverCnx, commandSender); + + // user receive the exception, send the same message again + // though we use the same TypedMessageBuilder, the two messages are different! + // because the sequence id is different, so the message is duplicated too. + typeMessages.sendAsync().exceptionally(e -> { + // should not enter here + Assert.fail(); + return null; + }).get(); + + // consume the message, there are two messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), 0); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), 1); + + // clean up + producer.close(); + consumer.close(); + } + From 28843c097ffceeea7294301c82c458793aad9eb5 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 24 Jul 2024 20:41:14 +0800 Subject: [PATCH 03/17] add test. --- .../client/api/DeduplicationEndToEndTest.java | 77 ++++++++++++++++++- 1 file changed, 73 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 0cbae87f878d0..1d4c375644ddc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -14,6 +14,7 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; /** * This test class is used to point out cases where message duplication can occur, @@ -106,8 +107,8 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { /** * simulate the case where the producer sends the message but doesn't receive the ack * due to network issue, broker issue, etc. User receives the exception and sends the same message again. - * With deduplication enabled, the message is duplicated too! Because the message newly sent has a - * different sequence id. + * With deduplication enabled, the message is duplicated too! Because the message newly sent by calling + * producer.sendAsync has a different sequence id. * @throws Exception */ @Test @@ -179,8 +180,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exceptio /** * simulate the case where the producer sends the message but doesn't receive the ack * due to network issue, broker issue, etc. User receives the exception and sends the same message again. - * With deduplication enabled, the message is duplicated too! Because the message newly sent has a - * different sequence id. + * With deduplication enabled, the message is duplicated too! Because the message newly sent by calling + * typeMessages.sendAsync() has a different sequence id, though we use the same TypedMessageBuilder. * @throws Exception */ @Test @@ -251,6 +252,74 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti } + /** + * simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. User receives the exception and sends the same message again. + * With deduplication enabled and user control sequence id, the message is not duplicated. + */ + @Test + public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceId() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id"; + admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); + + // Create producer with deduplication enabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic, false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // use reflection to spy the commandSender + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); + commandSenderField.set(serverCnx, spyCommandSender); + + // disable the send receipt + Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), + Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + + // send a message + long lastId = 0; + producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().thenRun(() -> { + // should not enter here + Assert.fail(); + }).exceptionally(e -> { + // do not receive the ack, should enter here + return null; + }).get(); + + + // set back the send receipt + commandSenderField.set(serverCnx, commandSender); + + // user receive the exception, send the same message again with the same sequence id. + producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().exceptionally(e -> { + // should not enter here + Assert.fail(); + return null; + }).get(); + + // consume the message, there are only one messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), lastId); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNull(message); + + // clean up + producer.close(); + consumer.close(); + } + } From fa1da3bcc867212b578d9991058c7bdedb26c999 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 25 Jul 2024 10:19:44 +0800 Subject: [PATCH 04/17] add test. --- .../client/api/DeduplicationEndToEndTest.java | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 1d4c375644ddc..cdb2fc545b6f9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -320,6 +320,77 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ consumer.close(); } + /** + * simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. User receives the exception and sends the same message again. + * With deduplication enabled and user control sequence id, but the topic is multi partitioned, + * the message is duplicated as message deduplication can't work across partitions. + */ + @Test + public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdPartitionedTopic() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-partitioned"; + admin.topics().createPartitionedTopic(topic, 2); + admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); + + // Create producer with deduplication enabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic + "-partition-0", false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // use reflection to spy the commandSender + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); + commandSenderField.set(serverCnx, spyCommandSender); + + // disable the send receipt + Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), + Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + + // send a message + long lastId = 0; + producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().thenRun(() -> { + // should not enter here + Assert.fail(); + }).exceptionally(e -> { + // do not receive the ack, should enter here + return null; + }).get(); + + // set back the send receipt + commandSenderField.set(serverCnx, commandSender); + + // user receive the exception, send the same message again with the same sequence id. + // but this new message will be routed to another partition, so the message is duplicated. + producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().exceptionally(e -> { + // should not enter here + Assert.fail(); + return null; + }).get(); + + // consume the message, there are two messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), lastId); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), lastId); + + // clean up + producer.close(); + consumer.close(); + } } From 498ec9e60592dcd1065714e75698f4b4a72f4144 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 25 Jul 2024 10:28:14 +0800 Subject: [PATCH 05/17] add code. --- .../client/api/DeduplicationEndToEndTest.java | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index cdb2fc545b6f9..1e9fb9c9f153c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -393,4 +393,75 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ } + /** + * simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. User receives the exception and sends the same message again. + * With deduplication enabled and user control sequence id, but the topic is multi partitioned, + * the message is duplicated as message deduplication can't work across partitions. + */ + @Test + public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdPartitionedTopicAndKeyBasedRouteProducer() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-partitioned-key-based-route-producer"; + admin.topics().createPartitionedTopic(topic, 2); + admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); + + // Create producer with deduplication enabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic + "-partition-0", false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // use reflection to spy the commandSender + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); + commandSenderField.set(serverCnx, spyCommandSender); + + // disable the send receipt + Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), + Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + + // send a message, with sequence id as the key, so messages with the same key will be routed to the same partition + long lastId = 0; + producer.newMessage().value("test".getBytes()).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().thenRun(() -> { + // should not enter here + Assert.fail(); + }).exceptionally(e -> { + // do not receive the ack, should enter here + return null; + }).get(); + + // set back the send receipt + commandSenderField.set(serverCnx, commandSender); + + // user receive the exception, send the same message again with the same sequence id. + // this new message will be routed to the same partition, so the message will not be duplicated. + producer.newMessage().value("test".getBytes()).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().exceptionally(e -> { + // should not enter here + Assert.fail(); + return null; + }).get(); + + // consume the message, there are two messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(new String(message.getData()), "test"); + assertEquals(message.getSequenceId(), lastId); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNull(message); + + // clean up + producer.close(); + consumer.close(); + } + + } From bd5375e8a50681425806708c0fbc8a67fd6c4fc8 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 25 Jul 2024 11:24:28 +0800 Subject: [PATCH 06/17] refactor. --- .../client/api/DeduplicationEndToEndTest.java | 222 ++++++++---------- 1 file changed, 95 insertions(+), 127 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 1e9fb9c9f153c..5f776b1656281 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -35,6 +35,46 @@ protected void cleanup() throws Exception { super.internalCleanup(); } + /** + * Disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. + * Multiple partitions use the same ServerCnx, so we need to disable the send receipt for one partition only. + * @param topic + * @param producerName + * @return + * @throws Exception + */ + private PulsarCommandSender disableSendReceipt(String topic, String producerName) throws Exception { + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic + "-partition-" + 0, false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // use reflection to spy the commandSender + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); + commandSenderField.set(serverCnx, spyCommandSender); + + // disable the send receipt + Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), + Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + return commandSender; + } + + private void enableSendReceipt(String topic, String producerName, PulsarCommandSender sender) throws Exception { + PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() + .getTopic(topic + "-partition-" + 0, false).get().get(); + assertNotNull(persistentTopic); + ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); + + // set original commandSender back + Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); + commandSenderField.setAccessible(true); + commandSenderField.set(serverCnx, sender); + } + /** * simulate the case where the producer sends the message but doesn't receive the ack * due to network issue, broker issue, etc. User receives the exception and sends the same message again. @@ -44,6 +84,8 @@ protected void cleanup() throws Exception { @Test public void testProducerDuplicationWithReceiptLost() throws Exception { final String topic = "persistent://my-property/my-ns/deduplication-test"; + int partitionCount = 1; + admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", false); // Create producer with deduplication enabled @@ -54,24 +96,11 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack - PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() - .getTopic(topic, false).get().get(); - assertNotNull(persistentTopic); - ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - - // use reflection to spy the commandSender - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); - PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); - commandSenderField.set(serverCnx, spyCommandSender); - - // disable the send receipt - Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), - Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + PulsarCommandSender sender = disableSendReceipt(topic, producerName); // send a message - producer.sendAsync("test".getBytes()).thenRun(() -> { + byte[] data = "test".getBytes(); + producer.sendAsync(data).thenRun(() -> { // should not enter here Assert.fail(); }).exceptionally(e -> { @@ -79,12 +108,11 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { return null; }).get(); - // set back the send receipt - commandSenderField.set(serverCnx, commandSender); + enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again - producer.sendAsync("test".getBytes()).exceptionally(e -> { + producer.sendAsync(data).exceptionally(e -> { // should not enter here Assert.fail(); return null; @@ -93,10 +121,10 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { // consume the message, there are two messages in the topic Message message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); // clean up producer.close(); @@ -114,6 +142,8 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { @Test public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exception { final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled"; + int partitionCount = 1; + admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); // Create producer with deduplication enabled @@ -124,24 +154,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exceptio Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack - PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() - .getTopic(topic, false).get().get(); - assertNotNull(persistentTopic); - ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - - // use reflection to spy the commandSender - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); - PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); - commandSenderField.set(serverCnx, spyCommandSender); - - // disable the send receipt - Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), - Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + PulsarCommandSender sender = disableSendReceipt(topic, producerName); // send a message - producer.sendAsync("test".getBytes()).thenRun(() -> { + byte[] data = "test".getBytes(); + producer.sendAsync(data).thenRun(() -> { // should not enter here Assert.fail(); }).exceptionally(e -> { @@ -149,13 +166,12 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exceptio return null; }).get(); - // set back the send receipt - commandSenderField.set(serverCnx, commandSender); + enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again // though the message content is the same, the sequence id is different, so the message is duplicated - producer.sendAsync("test".getBytes()).exceptionally(e -> { + producer.sendAsync(data).exceptionally(e -> { // should not enter here Assert.fail(); return null; @@ -164,11 +180,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exceptio // consume the message, there are two messages in the topic Message message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), 0); message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), 1); // clean up @@ -187,6 +203,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exceptio @Test public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Exception { final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled2"; + int partitionCount = 1; + admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); // Create producer with deduplication enabled @@ -197,24 +215,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack - PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() - .getTopic(topic, false).get().get(); - assertNotNull(persistentTopic); - ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - - // use reflection to spy the commandSender - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); - PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); - commandSenderField.set(serverCnx, spyCommandSender); - - // disable the send receipt - Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), - Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + PulsarCommandSender sender = disableSendReceipt(topic, producerName); // send a message - TypedMessageBuilder typeMessages = producer.newMessage().value("test".getBytes()); + byte[] data = "test".getBytes(); + TypedMessageBuilder typeMessages = producer.newMessage().value(data); typeMessages.sendAsync().thenRun(() -> { // should not enter here Assert.fail(); @@ -223,9 +228,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti return null; }).get(); - // set back the send receipt - commandSenderField.set(serverCnx, commandSender); + enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again // though we use the same TypedMessageBuilder, the two messages are different! @@ -239,11 +243,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti // consume the message, there are two messages in the topic Message message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), 0); message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), 1); // clean up @@ -260,6 +264,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti @Test public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceId() throws Exception { final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id"; + int partitionCount = 1; + admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); // Create producer with deduplication enabled @@ -270,25 +276,12 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack - PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() - .getTopic(topic, false).get().get(); - assertNotNull(persistentTopic); - ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - - // use reflection to spy the commandSender - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); - PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); - commandSenderField.set(serverCnx, spyCommandSender); - - // disable the send receipt - Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), - Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + PulsarCommandSender sender = disableSendReceipt(topic, producerName); // send a message long lastId = 0; - producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().thenRun(() -> { + byte[] data = "test".getBytes(); + producer.newMessage().value(data).sequenceId(lastId).sendAsync().thenRun(() -> { // should not enter here Assert.fail(); }).exceptionally(e -> { @@ -296,12 +289,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ return null; }).get(); - // set back the send receipt - commandSenderField.set(serverCnx, commandSender); + enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again with the same sequence id. - producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().exceptionally(e -> { + producer.newMessage().value(data).sequenceId(lastId).sendAsync().exceptionally(e -> { // should not enter here Assert.fail(); return null; @@ -310,7 +302,7 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // consume the message, there are only one messages in the topic Message message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), lastId); message = consumer.receive(1, TimeUnit.SECONDS); assertNull(message); @@ -327,9 +319,10 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ * the message is duplicated as message deduplication can't work across partitions. */ @Test - public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdPartitionedTopic() throws Exception { - final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-partitioned"; - admin.topics().createPartitionedTopic(topic, 2); + public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitioned() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-multi-partitioned"; + int partitionCount = 2; + admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); // Create producer with deduplication enabled @@ -340,25 +333,12 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack - PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() - .getTopic(topic + "-partition-0", false).get().get(); - assertNotNull(persistentTopic); - ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - - // use reflection to spy the commandSender - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); - PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); - commandSenderField.set(serverCnx, spyCommandSender); - - // disable the send receipt - Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), - Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + PulsarCommandSender sender = disableSendReceipt(topic, producerName); // send a message long lastId = 0; - producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().thenRun(() -> { + byte[] data = "test".getBytes(); + producer.newMessage().value(data).sequenceId(lastId).sendAsync().thenRun(() -> { // should not enter here Assert.fail(); }).exceptionally(e -> { @@ -367,11 +347,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ }).get(); // set back the send receipt - commandSenderField.set(serverCnx, commandSender); + enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again with the same sequence id. // but this new message will be routed to another partition, so the message is duplicated. - producer.newMessage().value("test".getBytes()).sequenceId(lastId).sendAsync().exceptionally(e -> { + producer.newMessage().value(data).sequenceId(lastId).sendAsync().exceptionally(e -> { // should not enter here Assert.fail(); return null; @@ -380,11 +360,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // consume the message, there are two messages in the topic Message message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), lastId); message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), lastId); // clean up @@ -400,9 +380,10 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ * the message is duplicated as message deduplication can't work across partitions. */ @Test - public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdPartitionedTopicAndKeyBasedRouteProducer() throws Exception { - final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-partitioned-key-based-route-producer"; - admin.topics().createPartitionedTopic(topic, 2); + public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitionedAndKeyBasedRouteProducer() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-multi-partitioned-key-based-route-producer"; + int partitionCount = 2; + admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); // Create producer with deduplication enabled @@ -413,25 +394,12 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack - PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() - .getTopic(topic + "-partition-0", false).get().get(); - assertNotNull(persistentTopic); - ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - - // use reflection to spy the commandSender - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); - PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); - commandSenderField.set(serverCnx, spyCommandSender); - - // disable the send receipt - Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), - Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong()); + PulsarCommandSender sender = disableSendReceipt(topic, producerName); // send a message, with sequence id as the key, so messages with the same key will be routed to the same partition long lastId = 0; - producer.newMessage().value("test".getBytes()).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().thenRun(() -> { + byte[] data = "test".getBytes(); + producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().thenRun(() -> { // should not enter here Assert.fail(); }).exceptionally(e -> { @@ -440,11 +408,11 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ }).get(); // set back the send receipt - commandSenderField.set(serverCnx, commandSender); + enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again with the same sequence id. // this new message will be routed to the same partition, so the message will not be duplicated. - producer.newMessage().value("test".getBytes()).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().exceptionally(e -> { + producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().exceptionally(e -> { // should not enter here Assert.fail(); return null; @@ -453,7 +421,7 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // consume the message, there are two messages in the topic Message message = consumer.receive(1, TimeUnit.SECONDS); assertNotNull(message); - assertEquals(new String(message.getData()), "test"); + assertEquals(message.getData(), data); assertEquals(message.getSequenceId(), lastId); message = consumer.receive(1, TimeUnit.SECONDS); assertNull(message); From c2bfc28635f9f9462b0c158c3a915951a394e0d5 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 25 Jul 2024 11:30:10 +0800 Subject: [PATCH 07/17] fix. --- .../pulsar/client/api/DeduplicationEndToEndTest.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 5f776b1656281..188a6cfe397bc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -63,6 +63,13 @@ private PulsarCommandSender disableSendReceipt(String topic, String producerName return commandSender; } + /** + * Set the original commandSender back to the ServerCnx, so that the producer can receive the ack. + * @param topic + * @param producerName + * @param sender + * @throws Exception + */ private void enableSendReceipt(String topic, String producerName, PulsarCommandSender sender) throws Exception { PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService() .getTopic(topic + "-partition-" + 0, false).get().get(); @@ -328,7 +335,7 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // Create producer with deduplication enabled String producerName = "my-producer-name"; Producer producer = pulsarClient.newProducer().topic(topic) - .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).enableBatching(false).create(); assertEquals(producer.getLastSequenceId(), -1L); Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); @@ -389,7 +396,7 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // Create producer with deduplication enabled String producerName = "my-producer-name"; Producer producer = pulsarClient.newProducer().topic(topic) - .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).enableBatching(false).create(); assertEquals(producer.getLastSequenceId(), -1L); Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); From 1010db7c5a8efdc630e492682e9d4c644d22d5f6 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 25 Jul 2024 12:10:06 +0800 Subject: [PATCH 08/17] add test. --- .../client/api/DeduplicationEndToEndTest.java | 114 +++++++++++++++++- 1 file changed, 112 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 188a6cfe397bc..1c27e91d65ab8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -1,8 +1,13 @@ package org.apache.pulsar.client.api; +import io.netty.util.TimerTask; import org.apache.pulsar.broker.service.PulsarCommandSender; import org.apache.pulsar.broker.service.ServerCnx; import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl; +import org.apache.pulsar.client.impl.PartitionedProducerImpl; +import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; import org.testng.annotations.AfterClass; @@ -383,8 +388,9 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ /** * simulate the case where the producer sends the message but doesn't receive the ack * due to network issue, broker issue, etc. User receives the exception and sends the same message again. - * With deduplication enabled and user control sequence id, but the topic is multi partitioned, - * the message is duplicated as message deduplication can't work across partitions. + * With deduplication enabled and user control sequence id, although the topic is multi partitioned, + * we use the key based routing to route the messages with the same key to the same partition. + * The message is not duplicated. */ @Test public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitionedAndKeyBasedRouteProducer() throws Exception { @@ -438,5 +444,109 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ consumer.close(); } + /** + * trigger the partition number update for the producer + * @param producer + * @param expectedPartitionCount + * @throws Exception + */ + private void triggerPartitionUpdateForPartitionedProducer(PartitionedProducerImpl producer, int expectedPartitionCount) throws Exception { + Field partitionsAutoUpdateTimerTask = PartitionedProducerImpl.class.getDeclaredField("partitionsAutoUpdateTimerTask"); + partitionsAutoUpdateTimerTask.setAccessible(true); + TimerTask timerTask = (TimerTask) partitionsAutoUpdateTimerTask.get(producer); + ((PulsarClientImpl) pulsarClient).getTimer().newTimeout(timerTask, 0, TimeUnit.MILLISECONDS); + Awaitility.await().until(() -> { + try { + return producer.getNumOfPartitions() == expectedPartitionCount; + } catch (Exception e) { + return false; + } + }); + } + + private void triggerPartitionUpdateForPartitionedConsumer(MultiTopicsConsumerImpl consumer, int expectedPartitionCount) throws Exception { + Field partitionsAutoUpdateTimerTask = MultiTopicsConsumerImpl.class.getDeclaredField("partitionsAutoUpdateTimerTask"); + partitionsAutoUpdateTimerTask.setAccessible(true); + TimerTask timerTask = (TimerTask) partitionsAutoUpdateTimerTask.get(consumer); + ((PulsarClientImpl) pulsarClient).getTimer().newTimeout(timerTask, 0, TimeUnit.MILLISECONDS); + Awaitility.await().until(() -> { + try { + return consumer.getPartitions().size() == expectedPartitionCount; + } catch (Exception e) { + return false; + } + }); + } + + /** + * simulate the case where the producer sends the message but doesn't receive the ack + * due to network issue, broker issue, etc. User receives the exception and sends the same message again. + * With deduplication enabled and user control sequence id, although the topic is multi partitioned, + * we use the key based routing to route the messages with the same key to the same partition. + * The message is not duplicated. + */ + @Test + public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitionedAndKeyBasedRouteProducerWhileUpdatePartition() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-multi-partitioned-key-based-route-producer-while-update-partition"; + int partitionCount = 2; + admin.topics().createPartitionedTopic(topic, partitionCount); + admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); + + // Create producer with deduplication enabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).enableBatching(false).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack + PulsarCommandSender sender = disableSendReceipt(topic, producerName); + + // send a message, with sequence id as the key, so messages with the same key will be routed to the same partition + long lastId = 0; + byte[] data = "test".getBytes(); + producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().thenRun(() -> { + // should not enter here + Assert.fail(); + }).exceptionally(e -> { + // do not receive the ack, should enter here + return null; + }).get(); + + // set back the send receipt + enableSendReceipt(topic, producerName, sender); + + // update the partition number between the two messages + admin.topics().updatePartitionedTopic(topic, 5); + + // trigger the partition number update for producer and consumer + triggerPartitionUpdateForPartitionedProducer((PartitionedProducerImpl) producer, 5); + triggerPartitionUpdateForPartitionedConsumer((MultiTopicsConsumerImpl) consumer, 5); + + // user receive the exception, send the same message again with the same sequence id. + // though the key of two messages are the same, the message is routed to different partition due to + // partition number update, so the message is duplicated. + producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().exceptionally(e -> { + // should not enter here + Assert.fail(); + return null; + }).get(); + + // consume the message, there are two messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + assertEquals(message.getSequenceId(), lastId); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + assertEquals(message.getSequenceId(), lastId); + + // clean up + producer.close(); + consumer.close(); + } + + } From eb24d815603b7efb67317cf66ce81b999f533803 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 10:02:32 +0800 Subject: [PATCH 09/17] add test. --- .../client/api/DeduplicationEndToEndTest.java | 106 +++++++++++++++++- 1 file changed, 105 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 1c27e91d65ab8..b13eeef5bc24a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -4,17 +4,24 @@ import org.apache.pulsar.broker.service.PulsarCommandSender; import org.apache.pulsar.broker.service.ServerCnx; import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.client.impl.ClientCnx; +import org.apache.pulsar.client.impl.ConnectionHandler; import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl; import org.apache.pulsar.client.impl.PartitionedProducerImpl; +import org.apache.pulsar.client.impl.ProducerImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import static org.testng.Assert.assertEquals; @@ -100,7 +107,7 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", false); - // Create producer with deduplication enabled + // Create producer with deduplication disabled String producerName = "my-producer-name"; Producer producer = pulsarClient.newProducer().topic(topic) .producerName(producerName).sendTimeout(1, TimeUnit.SECONDS).create(); @@ -547,6 +554,103 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ consumer.close(); } + /** + * trigger the reconnection + * @param producer + * @throws Exception + */ + private void triggerReconnection(PartitionedProducerImpl producer) throws Exception { + ProducerImpl producer1 = producer.getProducers().get(0); + ClientCnx cnx = producer1.getClientCnx(); + Field connectionHandlerField = ProducerImpl.class.getDeclaredField("connectionHandler"); + connectionHandlerField.setAccessible(true); + ConnectionHandler connectionHandler = (ConnectionHandler) connectionHandlerField.get(producer1); + connectionHandler.connectionClosed(cnx); + Awaitility.await().until(() -> { + try { + return connectionHandler.cnx() != null; + } catch (Exception e) { + return false; + } + }); + } + private void assertDuplicate(Consumer consumer, byte[] data) throws PulsarClientException { + // consume the message, there are at least two messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + } + + private void assertNotDuplicate(Consumer consumer, byte[] data) throws PulsarClientException { + // consume the message, there are only one messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNull(message); + } + + @DataProvider(name = "enableDedup") + public static Object[][] topicVersions() { + return new Object[][] { + { true }, + { false } + }; + } + + /** + * simulate the case when the connection is lost, producer resend the message internally with the same sequence id. + * If deduplication is not enabled, the message is duplicated. + * If deduplication is enabled, the message is not duplicated. + * @throws Exception + */ + @Test(dataProvider = "enableDedup") + public void testProducerDuplicationWhileReconnection(boolean enableDedup) throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-reconnection" + enableDedup; + int partitionCount = 1; + admin.topics().createPartitionedTopic(topic, partitionCount); + admin.namespaces().setDeduplicationStatus("my-property/my-ns", enableDedup); + + // Create producer with deduplication disabled + String producerName = "my-producer-name"; + Producer producer = pulsarClient.newProducer().topic(topic) + .producerName(producerName).sendTimeout(0, TimeUnit.SECONDS).enableBatching(false).create(); + assertEquals(producer.getLastSequenceId(), -1L); + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + + // disable the send receipt + PulsarCommandSender sender = disableSendReceipt(topic, producerName); + + // send a message + byte[] data = "test".getBytes(); + CompletableFuture sendFuture = producer.sendAsync(data); + producer.flushAsync(); + + // trigger reconnection before the producer receives the ack + // resend the message internally with the same sequence id + triggerReconnection((PartitionedProducerImpl) producer); + + // set back the send receipt + enableSendReceipt(topic, producerName, sender); + triggerReconnection((PartitionedProducerImpl) producer); + + sendFuture.get(5, TimeUnit.SECONDS); + + if (enableDedup) { + // with deduplication enabled, the message is not duplicated + assertNotDuplicate(consumer, data); + } else { + // with deduplication disabled, the message is duplicated + assertDuplicate(consumer, data); + } + + // clean up + producer.close(); + consumer.close(); + } } From 874d1f67d86ef9855abbf83ed71a112a24826783 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 10:17:31 +0800 Subject: [PATCH 10/17] refactor. --- .../client/api/DeduplicationEndToEndTest.java | 120 +++++++----------- 1 file changed, 47 insertions(+), 73 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index b13eeef5bc24a..08a1dd23b6ac1 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -19,8 +19,8 @@ import org.testng.annotations.Test; import java.lang.reflect.Field; -import java.lang.reflect.Method; -import java.util.Optional; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -47,6 +47,29 @@ protected void cleanup() throws Exception { super.internalCleanup(); } + private List> assertDuplicate(Consumer consumer, byte[] data) throws PulsarClientException { + // consume the message, there are at least two messages in the topic + List> messages = new ArrayList<>(2); + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + messages.add(message); + message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + messages.add(message); + return messages; + } + + private Message assertNotDuplicate(Consumer consumer, byte[] data) throws PulsarClientException { + // consume the message, there are only one messages in the topic + Message message = consumer.receive(1, TimeUnit.SECONDS); + assertNotNull(message); + assertEquals(message.getData(), data); + assertNull(consumer.receive(1, TimeUnit.SECONDS)); + return message; + } + /** * Disable the send receipt to simulate the case where the producer sends the message but doesn't receive the ack * due to network issue, broker issue, etc. @@ -138,12 +161,9 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { }).get(); // consume the message, there are two messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); + List> messages = assertDuplicate(consumer, data); + assertEquals(messages.get(0).getSequenceId(), 0); + assertEquals(messages.get(1).getSequenceId(), 1); // clean up producer.close(); @@ -197,14 +217,9 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exceptio }).get(); // consume the message, there are two messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), 0); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), 1); + List> messages = assertDuplicate(consumer, data); + assertEquals(messages.get(0).getSequenceId(), 0); + assertEquals(messages.get(1).getSequenceId(), 1); // clean up producer.close(); @@ -260,14 +275,9 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti }).get(); // consume the message, there are two messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), 0); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), 1); + List> messages = assertDuplicate(consumer, data); + assertEquals(messages.get(0).getSequenceId(), 0); + assertEquals(messages.get(1).getSequenceId(), 1); // clean up producer.close(); @@ -319,12 +329,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ }).get(); // consume the message, there are only one messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); + Message message = assertNotDuplicate(consumer, data); assertEquals(message.getSequenceId(), lastId); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNull(message); // clean up producer.close(); @@ -377,14 +383,9 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ }).get(); // consume the message, there are two messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), lastId); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), lastId); + List> messages = assertDuplicate(consumer, data); + assertEquals(messages.get(0).getSequenceId(), lastId); + assertEquals(messages.get(1).getSequenceId(), lastId); // clean up producer.close(); @@ -438,13 +439,9 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ return null; }).get(); - // consume the message, there are two messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); + // consume the message, there are only one messages in the topic + Message message = assertNotDuplicate(consumer, data); assertEquals(message.getSequenceId(), lastId); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNull(message); // clean up producer.close(); @@ -488,9 +485,10 @@ private void triggerPartitionUpdateForPartitionedConsumer(MultiTopicsConsumerImp /** * simulate the case where the producer sends the message but doesn't receive the ack * due to network issue, broker issue, etc. User receives the exception and sends the same message again. - * With deduplication enabled and user control sequence id, although the topic is multi partitioned, - * we use the key based routing to route the messages with the same key to the same partition. - * The message is not duplicated. + * With deduplication enabled and user control sequence id, the topic is multi partitioned, + * though we use the key based routing to route the messages with the same key to the same partition, + * If the partition number is updated between the two messages, the two messages will be routed to different partitions, + * so the message is duplicated. */ @Test public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitionedAndKeyBasedRouteProducerWhileUpdatePartition() throws Exception { @@ -540,14 +538,9 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ }).get(); // consume the message, there are two messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), lastId); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - assertEquals(message.getSequenceId(), lastId); + List> messages = assertDuplicate(consumer, data); + assertEquals(messages.get(0).getSequenceId(), lastId); + assertEquals(messages.get(1).getSequenceId(), lastId); // clean up producer.close(); @@ -575,25 +568,6 @@ private void triggerReconnection(PartitionedProducerImpl producer) throw }); } - private void assertDuplicate(Consumer consumer, byte[] data) throws PulsarClientException { - // consume the message, there are at least two messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - } - - private void assertNotDuplicate(Consumer consumer, byte[] data) throws PulsarClientException { - // consume the message, there are only one messages in the topic - Message message = consumer.receive(1, TimeUnit.SECONDS); - assertNotNull(message); - assertEquals(message.getData(), data); - message = consumer.receive(1, TimeUnit.SECONDS); - assertNull(message); - } - @DataProvider(name = "enableDedup") public static Object[][] topicVersions() { return new Object[][] { From a91966f6d773ac6ff94e2bf40aad900a4c1efae7 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 10:41:48 +0800 Subject: [PATCH 11/17] add code. --- .../client/api/DeduplicationEndToEndTest.java | 31 ++++++++++++++----- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 08a1dd23b6ac1..c2865aa6fde5d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -31,6 +31,22 @@ /** * This test class is used to point out cases where message duplication can occur, * producer idempotency features can be used to solve which cases and can't solve which cases. + * There are mainly two kinds of message duplication: + * 1. Producer sends the message but doesn't receive the ack due to network issue, broker issue, etc. + * User receives the exception and sends the same message again. + * This resend operation is executed by the user, so the user can control the sequence id of the message or not. + * Without message deduplication, the message is duplicated definitely. + * With message deduplication, the cases vary: + * - If user don't control the sequence id, the message is duplicated, as the message is resent with a different sequence id. + * - If user control the sequence id, there are chances that the message is duplicate or not, depending on whether + * the topic is single partitioned or multi partitioned, whether the sequence id is used as the key of the message, + * whether the partition number is updated between the two messages. + * + * 2. The connection between the producer and the broker is broken after the producer sends the message, + * but before the producer receives the ack. The producer reconnects to the broker and sends the same message again internally. + * In this case, the producer can't control the sequence id of the message. The resent message remains the same as the original message. + * Without message deduplication, the message is duplicated. + * With message deduplication, the message is not duplicated. */ public class DeduplicationEndToEndTest extends ProducerConsumerBase { @@ -570,9 +586,11 @@ private void triggerReconnection(PartitionedProducerImpl producer) throw @DataProvider(name = "enableDedup") public static Object[][] topicVersions() { - return new Object[][] { - { true }, - { false } + return new Object[][]{ + {true, 2}, + {false, 2}, + {true, 1}, + {false, 1}, }; } @@ -583,9 +601,8 @@ public static Object[][] topicVersions() { * @throws Exception */ @Test(dataProvider = "enableDedup") - public void testProducerDuplicationWhileReconnection(boolean enableDedup) throws Exception { - final String topic = "persistent://my-property/my-ns/deduplication-test-reconnection" + enableDedup; - int partitionCount = 1; + public void testProducerDuplicationWhileReconnection(boolean enableDedup, int partitionCount) throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-reconnection" + enableDedup + partitionCount; admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", enableDedup); @@ -612,7 +629,7 @@ public void testProducerDuplicationWhileReconnection(boolean enableDedup) throws enableSendReceipt(topic, producerName, sender); triggerReconnection((PartitionedProducerImpl) producer); - sendFuture.get(5, TimeUnit.SECONDS); + sendFuture.get(10, TimeUnit.SECONDS); if (enableDedup) { // with deduplication enabled, the message is not duplicated From 404e9e61fbb98cecb7bd2c606624b536c0713775 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 14:16:01 +0800 Subject: [PATCH 12/17] fix. --- .../client/api/DeduplicationEndToEndTest.java | 52 +++++++++++-------- 1 file changed, 29 insertions(+), 23 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index c2865aa6fde5d..50d35b2af7c79 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -45,8 +45,8 @@ * 2. The connection between the producer and the broker is broken after the producer sends the message, * but before the producer receives the ack. The producer reconnects to the broker and sends the same message again internally. * In this case, the producer can't control the sequence id of the message. The resent message remains the same as the original message. - * Without message deduplication, the message is duplicated. - * With message deduplication, the message is not duplicated. + * - Without message deduplication, the message is duplicated. + * - With message deduplication, the message is not duplicated. */ public class DeduplicationEndToEndTest extends ProducerConsumerBase { @@ -569,33 +569,36 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ * @throws Exception */ private void triggerReconnection(PartitionedProducerImpl producer) throws Exception { - ProducerImpl producer1 = producer.getProducers().get(0); - ClientCnx cnx = producer1.getClientCnx(); - Field connectionHandlerField = ProducerImpl.class.getDeclaredField("connectionHandler"); - connectionHandlerField.setAccessible(true); - ConnectionHandler connectionHandler = (ConnectionHandler) connectionHandlerField.get(producer1); - connectionHandler.connectionClosed(cnx); - Awaitility.await().until(() -> { - try { - return connectionHandler.cnx() != null; - } catch (Exception e) { - return false; - } - }); + for(ProducerImpl p : producer.getProducers()) { + p.getClientCnx().ctx().channel().close(); + ClientCnx cnx = p.getClientCnx(); + Field connectionHandlerField = ProducerImpl.class.getDeclaredField("connectionHandler"); + connectionHandlerField.setAccessible(true); + ConnectionHandler connectionHandler = (ConnectionHandler) connectionHandlerField.get(p); + connectionHandler.connectionClosed(cnx); + Awaitility.await().until(() -> { + try { + return connectionHandler.cnx() != null; + } catch (Exception e) { + return false; + } + }); + } } @DataProvider(name = "enableDedup") public static Object[][] topicVersions() { return new Object[][]{ - {true, 2}, {false, 2}, - {true, 1}, + {true, 2}, {false, 1}, + {true, 1}, }; } /** - * simulate the case when the connection is lost, producer resend the message internally with the same sequence id. + * simulate the case when the connection is lost, producer resend the message internally with the same sequence id + * to the same partition. * If deduplication is not enabled, the message is duplicated. * If deduplication is enabled, the message is not duplicated. * @throws Exception @@ -606,12 +609,12 @@ public void testProducerDuplicationWhileReconnection(boolean enableDedup, int pa admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", enableDedup); - // Create producer with deduplication disabled + // Create producer String producerName = "my-producer-name"; - Producer producer = pulsarClient.newProducer().topic(topic) + PartitionedProducerImpl producer = (PartitionedProducerImpl) pulsarClient.newProducer().topic(topic) .producerName(producerName).sendTimeout(0, TimeUnit.SECONDS).enableBatching(false).create(); assertEquals(producer.getLastSequenceId(), -1L); - Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); + admin.topics().createSubscription(topic, "my-sub", MessageId.earliest); // disable the send receipt PulsarCommandSender sender = disableSendReceipt(topic, producerName); @@ -623,14 +626,17 @@ public void testProducerDuplicationWhileReconnection(boolean enableDedup, int pa // trigger reconnection before the producer receives the ack // resend the message internally with the same sequence id - triggerReconnection((PartitionedProducerImpl) producer); + triggerReconnection(producer); // set back the send receipt enableSendReceipt(topic, producerName, sender); - triggerReconnection((PartitionedProducerImpl) producer); + triggerReconnection(producer); sendFuture.get(10, TimeUnit.SECONDS); + // create consumer after the reconnection, because reconnection will trigger the same message to be + // delivered to the consumer again, which is a duplication problem in consumer side. + Consumer consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("my-sub").subscribe(); if (enableDedup) { // with deduplication enabled, the message is not duplicated assertNotDuplicate(consumer, data); From 26b2a0c3503296897747fc4e65d6bc437d6f7b73 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 14:37:59 +0800 Subject: [PATCH 13/17] add test group. --- .../client/api/DeduplicationEndToEndTest.java | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 50d35b2af7c79..f7d91b21fc291 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -1,6 +1,25 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ package org.apache.pulsar.client.api; import io.netty.util.TimerTask; +import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.service.PulsarCommandSender; import org.apache.pulsar.broker.service.ServerCnx; import org.apache.pulsar.broker.service.persistent.PersistentTopic; @@ -48,6 +67,7 @@ * - Without message deduplication, the message is duplicated. * - With message deduplication, the message is not duplicated. */ +@Test(groups = "broker-api") public class DeduplicationEndToEndTest extends ProducerConsumerBase { @BeforeClass From 8dcf84ed25f4bf51036b4aded901bd10ea5e1b35 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 14:39:17 +0800 Subject: [PATCH 14/17] fix. --- .../org/apache/pulsar/client/api/DeduplicationEndToEndTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index f7d91b21fc291..76c2564580a35 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -19,7 +19,6 @@ package org.apache.pulsar.client.api; import io.netty.util.TimerTask; -import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.service.PulsarCommandSender; import org.apache.pulsar.broker.service.ServerCnx; import org.apache.pulsar.broker.service.persistent.PersistentTopic; From 2d7fc300fb823b288ed60b24876240ec29e622f7 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 16:09:46 +0800 Subject: [PATCH 15/17] refactor. --- .../client/api/DeduplicationEndToEndTest.java | 102 ++++-------------- 1 file changed, 18 insertions(+), 84 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 76c2564580a35..9d85230e16136 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -177,23 +177,13 @@ public void testProducerDuplicationWithReceiptLost() throws Exception { // send a message byte[] data = "test".getBytes(); - producer.sendAsync(data).thenRun(() -> { - // should not enter here - Assert.fail(); - }).exceptionally(e -> { - // do not receive the ack, should enter here - return null; - }).get(); + Assert.assertThrows(PulsarClientException.TimeoutException.class, () -> producer.send(data)); // set back the send receipt enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again - producer.sendAsync(data).exceptionally(e -> { - // should not enter here - Assert.fail(); - return null; - }).get(); + producer.send(data); // consume the message, there are two messages in the topic List> messages = assertDuplicate(consumer, data); @@ -232,24 +222,14 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled() throws Exceptio // send a message byte[] data = "test".getBytes(); - producer.sendAsync(data).thenRun(() -> { - // should not enter here - Assert.fail(); - }).exceptionally(e -> { - // do not receive the ack, should enter here - return null; - }).get(); + Assert.assertThrows(PulsarClientException.TimeoutException.class, () -> producer.send(data)); // set back the send receipt enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again // though the message content is the same, the sequence id is different, so the message is duplicated - producer.sendAsync(data).exceptionally(e -> { - // should not enter here - Assert.fail(); - return null; - }).get(); + producer.send(data); // consume the message, there are two messages in the topic List> messages = assertDuplicate(consumer, data); @@ -289,13 +269,7 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti // send a message byte[] data = "test".getBytes(); TypedMessageBuilder typeMessages = producer.newMessage().value(data); - typeMessages.sendAsync().thenRun(() -> { - // should not enter here - Assert.fail(); - }).exceptionally(e -> { - // do not receive the ack, should enter here - return null; - }).get(); + Assert.assertThrows(PulsarClientException.TimeoutException.class, () -> typeMessages.send()); // set back the send receipt enableSendReceipt(topic, producerName, sender); @@ -303,11 +277,7 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti // user receive the exception, send the same message again // though we use the same TypedMessageBuilder, the two messages are different! // because the sequence id is different, so the message is duplicated too. - typeMessages.sendAsync().exceptionally(e -> { - // should not enter here - Assert.fail(); - return null; - }).get(); + typeMessages.send(); // consume the message, there are two messages in the topic List> messages = assertDuplicate(consumer, data); @@ -345,23 +315,14 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // send a message long lastId = 0; byte[] data = "test".getBytes(); - producer.newMessage().value(data).sequenceId(lastId).sendAsync().thenRun(() -> { - // should not enter here - Assert.fail(); - }).exceptionally(e -> { - // do not receive the ack, should enter here - return null; - }).get(); + Assert.assertThrows(PulsarClientException.TimeoutException.class, + () -> producer.newMessage().value(data).sequenceId(lastId).send()); // set back the send receipt enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again with the same sequence id. - producer.newMessage().value(data).sequenceId(lastId).sendAsync().exceptionally(e -> { - // should not enter here - Assert.fail(); - return null; - }).get(); + producer.newMessage().value(data).sequenceId(lastId).send(); // consume the message, there are only one messages in the topic Message message = assertNotDuplicate(consumer, data); @@ -398,24 +359,15 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // send a message long lastId = 0; byte[] data = "test".getBytes(); - producer.newMessage().value(data).sequenceId(lastId).sendAsync().thenRun(() -> { - // should not enter here - Assert.fail(); - }).exceptionally(e -> { - // do not receive the ack, should enter here - return null; - }).get(); + Assert.assertThrows(PulsarClientException.TimeoutException.class, + () -> producer.newMessage().value(data).sequenceId(lastId).send()); // set back the send receipt enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again with the same sequence id. // but this new message will be routed to another partition, so the message is duplicated. - producer.newMessage().value(data).sequenceId(lastId).sendAsync().exceptionally(e -> { - // should not enter here - Assert.fail(); - return null; - }).get(); + producer.newMessage().value(data).sequenceId(lastId).send(); // consume the message, there are two messages in the topic List> messages = assertDuplicate(consumer, data); @@ -455,24 +407,15 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // send a message, with sequence id as the key, so messages with the same key will be routed to the same partition long lastId = 0; byte[] data = "test".getBytes(); - producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().thenRun(() -> { - // should not enter here - Assert.fail(); - }).exceptionally(e -> { - // do not receive the ack, should enter here - return null; - }).get(); + Assert.assertThrows(PulsarClientException.TimeoutException.class, + () -> producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).send()); // set back the send receipt enableSendReceipt(topic, producerName, sender); // user receive the exception, send the same message again with the same sequence id. // this new message will be routed to the same partition, so the message will not be duplicated. - producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().exceptionally(e -> { - // should not enter here - Assert.fail(); - return null; - }).get(); + producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).send(); // consume the message, there are only one messages in the topic Message message = assertNotDuplicate(consumer, data); @@ -545,13 +488,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // send a message, with sequence id as the key, so messages with the same key will be routed to the same partition long lastId = 0; byte[] data = "test".getBytes(); - producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().thenRun(() -> { - // should not enter here - Assert.fail(); - }).exceptionally(e -> { - // do not receive the ack, should enter here - return null; - }).get(); + Assert.assertThrows(PulsarClientException.TimeoutException.class, + () -> producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).send()); // set back the send receipt enableSendReceipt(topic, producerName, sender); @@ -566,11 +504,7 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ // user receive the exception, send the same message again with the same sequence id. // though the key of two messages are the same, the message is routed to different partition due to // partition number update, so the message is duplicated. - producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).sendAsync().exceptionally(e -> { - // should not enter here - Assert.fail(); - return null; - }).get(); + producer.newMessage().value(data).sequenceId(lastId).key(String.valueOf(lastId)).send(); // consume the message, there are two messages in the topic List> messages = assertDuplicate(consumer, data); From 62be23f68645fd4296aefe07117dfecc79f4d400 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 16:23:59 +0800 Subject: [PATCH 16/17] add setCommandSender --- .../apache/pulsar/broker/service/ServerCnx.java | 5 +++++ .../client/api/DeduplicationEndToEndTest.java | 16 +++++----------- 2 files changed, 10 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 6690ab4af5fd1..552331d8a334d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -3551,6 +3551,11 @@ public PulsarCommandSender getCommandSender() { return commandSender; } + @VisibleForTesting + public void setCommandSender(PulsarCommandSender commandSender) { + this.commandSender = commandSender; + } + @Override public void execute(Runnable runnable) { ctx().channel().eventLoop().execute(runnable); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 9d85230e16136..8a7d3361afeec 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -111,7 +111,7 @@ private Message assertNotDuplicate(Consumer consumer, byte[] dat * Multiple partitions use the same ServerCnx, so we need to disable the send receipt for one partition only. * @param topic * @param producerName - * @return + * @return the original commandSender * @throws Exception */ private PulsarCommandSender disableSendReceipt(String topic, String producerName) throws Exception { @@ -120,12 +120,9 @@ private PulsarCommandSender disableSendReceipt(String topic, String producerName assertNotNull(persistentTopic); ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - // use reflection to spy the commandSender - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - PulsarCommandSender commandSender = (PulsarCommandSender) commandSenderField.get(serverCnx); + PulsarCommandSender commandSender = serverCnx.getCommandSender(); PulsarCommandSender spyCommandSender = Mockito.spy(commandSender); - commandSenderField.set(serverCnx, spyCommandSender); + serverCnx.setCommandSender(spyCommandSender); // disable the send receipt Mockito.doNothing().when(spyCommandSender).sendSendReceiptResponse(Mockito.anyLong(), Mockito.anyLong(), @@ -137,7 +134,7 @@ private PulsarCommandSender disableSendReceipt(String topic, String producerName * Set the original commandSender back to the ServerCnx, so that the producer can receive the ack. * @param topic * @param producerName - * @param sender + * @param sender the original commandSender * @throws Exception */ private void enableSendReceipt(String topic, String producerName, PulsarCommandSender sender) throws Exception { @@ -145,11 +142,8 @@ private void enableSendReceipt(String topic, String producerName, PulsarCommandS .getTopic(topic + "-partition-" + 0, false).get().get(); assertNotNull(persistentTopic); ServerCnx serverCnx = (ServerCnx) persistentTopic.getProducers().get(producerName).getCnx(); - // set original commandSender back - Field commandSenderField = ServerCnx.class.getDeclaredField("commandSender"); - commandSenderField.setAccessible(true); - commandSenderField.set(serverCnx, sender); + serverCnx.setCommandSender(sender); } /** From 9244286b18bc8c6c5a45bfa29e456ab8f5961e5c Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 26 Jul 2024 16:33:01 +0800 Subject: [PATCH 17/17] shorten method name. --- .../client/api/DeduplicationEndToEndTest.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java index 8a7d3361afeec..d9b69a91db0a1 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeduplicationEndToEndTest.java @@ -290,8 +290,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabled2() throws Excepti * With deduplication enabled and user control sequence id, the message is not duplicated. */ @Test - public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceId() throws Exception { - final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id"; + public void testProducerDuplicationWithReceiptLostUserControlSequenceId() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-user-control-sequence-id"; int partitionCount = 1; admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); @@ -334,8 +334,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ * the message is duplicated as message deduplication can't work across partitions. */ @Test - public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitioned() throws Exception { - final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-multi-partitioned"; + public void testProducerDuplicationWithReceiptLostUserControlSequenceIdMultiPartitioned() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-user-control-sequence-id-multi-partitioned"; int partitionCount = 2; admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); @@ -382,8 +382,8 @@ public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequ * The message is not duplicated. */ @Test - public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitionedAndKeyBasedRouteProducer() throws Exception { - final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-multi-partitioned-key-based-route-producer"; + public void testProducerDuplicationWithReceiptLostUserControlSequenceIdKeyBasedRoute() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-user-control-sequence-id-key-based-route"; int partitionCount = 2; admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true); @@ -463,8 +463,8 @@ private void triggerPartitionUpdateForPartitionedConsumer(MultiTopicsConsumerImp * so the message is duplicated. */ @Test - public void testProducerDuplicationWithReceiptLostDedupEnabledAndUserControlSequenceIdMultiPartitionedAndKeyBasedRouteProducerWhileUpdatePartition() throws Exception { - final String topic = "persistent://my-property/my-ns/deduplication-test-dedup-enabled-user-control-sequence-id-multi-partitioned-key-based-route-producer-while-update-partition"; + public void testProducerDuplicationWithReceiptLostUserControlSequenceIdKeyBasedRouteWhileUpdatePartition() throws Exception { + final String topic = "persistent://my-property/my-ns/deduplication-test-user-control-sequence-id-key-based-route-while-update-partition"; int partitionCount = 2; admin.topics().createPartitionedTopic(topic, partitionCount); admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);