diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java index de00379bdb3a2..cd89fcc00f369 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java @@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit; import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; public class RetryTopicTest extends ProducerConsumerBase { @@ -65,7 +66,7 @@ public void testRetryTopic() throws Exception { PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection Consumer deadLetterConsumer = newPulsarClient.newConsumer(Schema.BYTES) - .topic("persistent://my-property/my-ns/my-subscription-DLQ") + .topic("persistent://my-property/my-ns/retry-topic-my-subscription-DLQ") .subscriptionName("my-subscription") .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) .subscribe(); @@ -82,7 +83,7 @@ public void testRetryTopic() throws Exception { int totalReceived = 0; do { - Message message = consumer.receive(); + Message message = consumer.receive(3, TimeUnit.SECONDS); log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData())); consumer.reconsumeLater(message, 1 , TimeUnit.SECONDS); totalReceived++; @@ -122,16 +123,27 @@ public void testRetryTopic() throws Exception { */ @Test public void testRetryTopicWithMultiTopic() throws Exception { - final String topic1 = "persistent://my-property/my-ns/retry-topic-1"; - final String topic2 = "persistent://my-property/my-ns/retry-topic-2"; + final String topic1 = "persistent://my-property/my-ns/topic-1"; + final String topic2 = "persistent://my-property/my-ns/topic-2"; final int maxRedeliveryCount = 2; - int sendMessages = 100; + int sendMessages = 10; // subscribe to the original topics before publish - Consumer consumer = pulsarClient.newConsumer(Schema.BYTES) - .topic(topic1, topic2) + Consumer consumer1 = pulsarClient.newConsumer(Schema.BYTES) + .topic(topic1) + .subscriptionName("my-subscription") + .subscriptionType(SubscriptionType.Shared) + .enableRetry(true) + .ackTimeout(1, TimeUnit.SECONDS) + .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).build()) + .receiverQueueSize(100) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Consumer consumer2 = pulsarClient.newConsumer(Schema.BYTES) + .topic(topic2) .subscriptionName("my-subscription") .subscriptionType(SubscriptionType.Shared) .enableRetry(true) @@ -142,8 +154,15 @@ public void testRetryTopicWithMultiTopic() throws Exception { .subscribe(); // subscribe to the DLQ topics before consuming original topics - Consumer deadLetterConsumer = pulsarClient.newConsumer(Schema.BYTES) - .topic("persistent://my-property/my-ns/my-subscription-DLQ") + Consumer deadLetterConsumer1 = pulsarClient.newConsumer(Schema.BYTES) + .topic("persistent://my-property/my-ns/topic-1-my-subscription-DLQ") + .subscriptionName("my-subscription") + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + // subscribe to the DLQ topics before consuming original topics + Consumer deadLetterConsumer2 = pulsarClient.newConsumer(Schema.BYTES) + .topic("persistent://my-property/my-ns/topic-2-my-subscription-DLQ") .subscriptionName("my-subscription") .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) .subscribe(); @@ -157,32 +176,51 @@ public void testRetryTopicWithMultiTopic() throws Exception { .create(); for (int i = 0; i < sendMessages; i++) { - producer1.send(String.format("Hello Pulsar [%d]", i).getBytes()); - producer2.send(String.format("Hello Pulsar [%d]", i).getBytes()); + producer1.send(String.format("Producer1 Hello Pulsar [%d]", i).getBytes()); + producer2.send(String.format("Producer2 Hello Pulsar [%d]", i).getBytes()); } - sendMessages = sendMessages * 2; - producer1.close(); producer2.close(); - int totalReceived = 0; + int totalReceivedConsumer1 = 0; do { - Message message = consumer.receive(); - log.info("consumer received message : {} {} - total = {}", - message.getMessageId(), new String(message.getData()), ++totalReceived); - } while (totalReceived < sendMessages * (maxRedeliveryCount + 1)); + Message message = consumer1.receive(3, TimeUnit.SECONDS); + log.info("consumer1 received message : {} {} - total = {}", + message.getMessageId(), new String(message.getData()), ++totalReceivedConsumer1); + } while (totalReceivedConsumer1 < sendMessages * (maxRedeliveryCount + 1)); - int totalInDeadLetter = 0; + int totalReceivedConsumer2 = 0; do { - Message message = deadLetterConsumer.receive(); - log.info("dead letter consumer received message : {} {}", message.getMessageId(), new String(message.getData())); - deadLetterConsumer.acknowledge(message); - totalInDeadLetter++; - } while (totalInDeadLetter < sendMessages); + Message message = consumer2.receive(3, TimeUnit.SECONDS); + log.info("consumer2 received message : {} {} - total = {}", + message.getMessageId(), new String(message.getData()), ++totalReceivedConsumer2); + } while (totalReceivedConsumer2 < sendMessages * (maxRedeliveryCount + 1)); - deadLetterConsumer.close(); - consumer.close(); + int totalInDeadLetter1 = 0; + do { + Message message = deadLetterConsumer1.receive(3, TimeUnit.SECONDS); + log.info("dead letter consumer1 received message : {} {}", message.getMessageId(), new String(message.getData())); + // only receive message publish by producer1 + assertTrue(new String(message.getData()).startsWith("Producer1")); + deadLetterConsumer1.acknowledge(message); + totalInDeadLetter1++; + } while (totalInDeadLetter1 < sendMessages); + + int totalInDeadLetter2 = 0; + do { + Message message = deadLetterConsumer2.receive(3, TimeUnit.SECONDS); + log.info("dead letter consumer2 received message : {} {}", message.getMessageId(), new String(message.getData())); + // only receive message publish by producer2 + assertTrue(new String(message.getData()).startsWith("Producer2")); + deadLetterConsumer2.acknowledge(message); + totalInDeadLetter2++; + } while (totalInDeadLetter2 < sendMessages); + + deadLetterConsumer1.close(); + deadLetterConsumer2.close(); + consumer1.close(); + consumer2.close(); Consumer checkConsumer = pulsarClient.newConsumer(Schema.BYTES) .topic(topic1, topic2) @@ -221,7 +259,7 @@ public void testRetryTopicByCustomTopicName() throws Exception { .subscribe(); PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection Consumer deadLetterConsumer = newPulsarClient.newConsumer(Schema.BYTES) - .topic("persistent://my-property/my-ns/my-subscription-DLQ") + .topic("persistent://my-property/my-ns/retry-topic-my-subscription-DLQ") .subscriptionName("my-subscription") .subscribe(); @@ -235,20 +273,23 @@ public void testRetryTopicByCustomTopicName() throws Exception { int totalReceived = 0; do { - Message message = consumer.receive(); + Message message = consumer.receive(3, TimeUnit.SECONDS); log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData())); consumer.reconsumeLater(message, 1 , TimeUnit.SECONDS); totalReceived++; } while (totalReceived < sendMessages * (maxRedeliveryCount + 1)); + int totalInDeadLetter = 0; do { - Message message = deadLetterConsumer.receive(); + Message message = deadLetterConsumer.receive(3, TimeUnit.SECONDS); log.info("dead letter consumer received message : {} {}", message.getMessageId(), new String(message.getData())); deadLetterConsumer.acknowledge(message); totalInDeadLetter++; } while (totalInDeadLetter < sendMessages); + deadLetterConsumer.close(); consumer.close(); + PulsarClient newPulsarClient1 = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection Consumer checkConsumer = newPulsarClient1.newConsumer(Schema.BYTES) .topic(topic) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java index 95efe28767146..0128a45cbd845 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java @@ -358,7 +358,7 @@ public interface ConsumerBuilder extends Cloneable { * will be more efficient at the expense of a slight increase in message re-deliveries after a failure. * * @param delay - * the max amount of time an acknowledgemnt can be delayed + * the max amount of time an acknowledgement can be delayed * @param unit * the time unit for the delay * @return the consumer builder instance @@ -372,7 +372,7 @@ public interface ConsumerBuilder extends Cloneable { ConsumerBuilder replicateSubscriptionState(boolean replicateSubscriptionState); /** - * Set the max total receiver queue size across partitons. + * Set the max total receiver queue size across partitions. * *

This setting will be used to reduce the receiver queue size for individual partitions * {@link #receiverQueueSize(int)} if the total exceeds this value (default: 50000). @@ -389,7 +389,7 @@ public interface ConsumerBuilder extends Cloneable { /** * Set the consumer name. * - *

Consumer name is informative and it can be used to indentify a particular consumer + *

Consumer name is informative and it can be used to identify a particular consumer * instance from the topic stats. * * @param consumerName @@ -648,7 +648,7 @@ public interface ConsumerBuilder extends Cloneable { /** * If enabled, the consumer will auto retry message. - * default unabled. + * default disabled. * * @param retryEnable * whether to auto retry message diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java index 46bfb7997b954..554543e747189 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java @@ -119,10 +119,12 @@ public CompletableFuture> subscribeAsync() { return FutureUtil.failedFuture( new InvalidConfigurationException("KeySharedPolicy must set with KeyShared subscription")); } - if(conf.isRetryEnable() && conf.getTopicNames().size() > 0 ) { - TopicName topicFirst = TopicName.get(conf.getTopicNames().iterator().next()); - String retryLetterTopic = topicFirst.getNamespace() + "/" + conf.getSubscriptionName() + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX; - String deadLetterTopic = topicFirst.getNamespace() + "/" + conf.getSubscriptionName() + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX; + if (conf.isRetryEnable() && conf.getTopicNames().size() > 0 ) { + String topicName = TopicName.get(conf.getTopicNames().iterator().next()).getPartitionedTopicName(); + String retryLetterTopic = topicName + "-" + conf.getSubscriptionName() + + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX; + String deadLetterTopic = topicName + "-" + conf.getSubscriptionName() + + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX; if(conf.getDeadLetterPolicy() == null) { conf.setDeadLetterPolicy(DeadLetterPolicy.builder() .maxRedeliverCount(RetryMessageUtil.MAX_RECONSUMETIMES)