Skip to content

fix consumer receives duplicate messages when enable retry letter topic #9743

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
aloyszhang wants to merge 2 commits into from
Closed

fix consumer receives duplicate messages when enable retry letter topic #9743

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
be11af4
fix consumer receives duplicate messages when enable retry letter topic
aloyszhang Feb 26, 2021
89b7326
fix checkstyle problem
aloyszhang Feb 26, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 70 additions & 29 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand Down Expand Up @@ -65,7 +66,7 @@ public void testRetryTopic() throws Exception {

PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
Consumer<byte[]> 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();
Expand All @@ -82,7 +83,7 @@ public void testRetryTopic() throws Exception {

int totalReceived = 0;
do {
Message<byte[]> message = consumer.receive();
Message<byte[]> message = consumer.receive(3, TimeUnit.SECONDS);
log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
consumer.reconsumeLater(message, 1 , TimeUnit.SECONDS);
totalReceived++;
Expand Down Expand Up @@ -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<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
.topic(topic1, topic2)
Consumer<byte[]> 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<byte[]> consumer2 = pulsarClient.newConsumer(Schema.BYTES)
.topic(topic2)
.subscriptionName("my-subscription")
.subscriptionType(SubscriptionType.Shared)
.enableRetry(true)
Expand All @@ -142,8 +154,15 @@ public void testRetryTopicWithMultiTopic() throws Exception {
.subscribe();

// subscribe to the DLQ topics before consuming original topics
Consumer<byte[]> deadLetterConsumer = pulsarClient.newConsumer(Schema.BYTES)
.topic("persistent://my-property/my-ns/my-subscription-DLQ")
Consumer<byte[]> 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<byte[]> deadLetterConsumer2 = pulsarClient.newConsumer(Schema.BYTES)
.topic("persistent://my-property/my-ns/topic-2-my-subscription-DLQ")
.subscriptionName("my-subscription")
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.subscribe();
Expand All @@ -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<byte[]> message = consumer.receive();
log.info("consumer received message : {} {} - total = {}",
message.getMessageId(), new String(message.getData()), ++totalReceived);
} while (totalReceived < sendMessages * (maxRedeliveryCount + 1));
Message<byte[]> 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<byte[]> 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<byte[]> checkConsumer = pulsarClient.newConsumer(Schema.BYTES)
.topic(topic1, topic2)
Expand Down Expand Up @@ -221,7 +259,7 @@ public void testRetryTopicByCustomTopicName() throws Exception {
.subscribe();
PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
Consumer<byte[]> 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();

Expand All @@ -235,20 +273,23 @@ public void testRetryTopicByCustomTopicName() throws Exception {

int totalReceived = 0;
do {
Message<byte[]> message = consumer.receive();
Message<byte[]> 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<byte[]> checkConsumer = newPulsarClient1.newConsumer(Schema.BYTES)
.topic(topic)
Expand Down
8 changes: 4 additions & 4 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -358,7 +358,7 @@ public interface ConsumerBuilder<T> 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
Expand All @@ -372,7 +372,7 @@ public interface ConsumerBuilder<T> extends Cloneable {
ConsumerBuilder<T> replicateSubscriptionState(boolean replicateSubscriptionState);

/**
* Set the max total receiver queue size across partitons.
* Set the max total receiver queue size across partitions.
*
* <p>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).
Expand All @@ -389,7 +389,7 @@ public interface ConsumerBuilder<T> extends Cloneable {
/**
* Set the consumer name.
*
* <p>Consumer name is informative and it can be used to indentify a particular consumer
* <p>Consumer name is informative and it can be used to identify a particular consumer
* instance from the topic stats.
*
* @param consumerName
Expand Down Expand Up @@ -648,7 +648,7 @@ public interface ConsumerBuilder<T> extends Cloneable {

/**
* If enabled, the consumer will auto retry message.
* default unabled.
* default disabled.
*
* @param retryEnable
* whether to auto retry message
Expand Down
10 changes: 6 additions & 4 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,10 +119,12 @@ public CompletableFuture<Consumer<T>> 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;
Comment on lines +122 to +127

@codelipenghui codelipenghui Mar 1, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Topics might not belong to a partitioned topic since we can use .topic("persistent://my-tenant/my-ns/topic-a", "persistent://my-tenant/my-ns/topic-b", "persistent://my-tenant/my-ns/topic-c"). This is why the default retry topic name does not contains the topic(the topic name is written to the properties of the message when send it to the retry topic).

You can explicitly specify the retry topic name or the dead letter topic name through the dead letter policy.

@aloyszhang aloyszhang Mar 2, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for your explain. I got that why the original default retry letter topic does not contain the topic. There is no doubt that the best way to avoid this problem is specify the retry letter topic and dead letter topic.

In our prduction environment, we maintain a public pulsar cluster and many users access it. There are some users just set enableRetry to true without specify retry letter topic and dead letter topic which may lead to problem discribed by #9742.

IMO, it's better we can find a way to fix this problem. Maybe we can handle default retry and dead letter topic for each topic.

Topics might not belong to a partitioned topic since we can use .topic("persistent://my-tenant/my-ns/topic-a", "persistent://my-tenant/my-ns/topic-b", "persistent://my-tenant/my-ns/topic-c").

when consumer subscribe multi topics, pulsar can set the default retry letter topic and dead letter topic for every topic seperatlly, e.g.
for persistent://my-tenant/my-ns/topic-a will have
default retry letter topic persistent://my-tenant/my-ns/topic-a-subscription-name-RETRY
and dead letter topic persistent://my-tenant/my-ns/topic-a-subscription-name-DLQ;
for persistent://my-tenant/my-ns/topic-b will have
default retry letter topic persistent://my-tenant/my-ns/topic-b-subscription-name-RETRY
and dead letter topic persistent://my-tenant/my-ns/topic-b-subscription-name-DLQ

Without specify the dead letter topic, user should subscribe all default dead letter topic when receive message from DLQ.

@codelipenghui codelipenghui Mar 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Make sense to me. And I think after this change will break the users that uses the try letter topic due to the different topic name. So this should discuss it in the dev email channel and highlight this will be a breaking change, so that we can highlight it at the release note.

if(conf.getDeadLetterPolicy() == null) {
conf.setDeadLetterPolicy(DeadLetterPolicy.builder()
.maxRedeliverCount(RetryMessageUtil.MAX_RECONSUMETIMES)
Expand Down