You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2022/08/28 12:29:57 UTC
[pulsar] branch branch-2.10 updated: [fix][client] Fix reach redeliverCount client can't send batch messages to DLQ (#17317)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 718068d5bed [fix][client] Fix reach redeliverCount client can't send batch messages to DLQ (#17317)
718068d5bed is described below
commit 718068d5bedd7cf07d2d8e06efbc67f8d581e82a
Author: congbo <39...@users.noreply.github.com>
AuthorDate: Sun Aug 28 20:10:57 2022 +0800
[fix][client] Fix reach redeliverCount client can't send batch messages to DLQ (#17317)
(cherry picked from commit 0909853873fc61395df7a68de13335d9f770383a)
---
.../client/impl/TransactionEndToEndTest.java | 66 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 11 +++-
2 files changed, 75 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
index 1df364f18b2..6868f49fa14 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java
@@ -1116,6 +1116,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
@Cleanup
ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
.topic(topic)
+ .enableBatching(false)
.sendTimeout(1, TimeUnit.SECONDS)
.create();
@@ -1160,4 +1161,69 @@ public class TransactionEndToEndTest extends TransactionTestBase {
assertEquals(value, new String(deadLetterConsumer.receive(3, TimeUnit.SECONDS).getValue()));
}
+
+ @Test
+ public void testSendTxnAckBatchMessageToDLQ() throws Exception {
+ String topic = NAMESPACE1 + "/testSendTxnAckBatchMessageToDLQ";
+ String subName = "test";
+ String value1 = "test1";
+ String value2 = "test2";
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .sendTimeout(1, TimeUnit.SECONDS)
+ .create();
+
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionType(SubscriptionType.Shared)
+ // consumer can't receive the same message three times
+ .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(1).build())
+ .subscriptionName(subName)
+ .subscribe();
+
+ @Cleanup
+ Consumer<byte[]> deadLetterConsumer = pulsarClient.newConsumer()
+ .topic(String.format("%s-%s" + RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX,
+ topic, subName))
+ .subscriptionType(SubscriptionType.Shared)
+ .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(1).build())
+ .subscriptionName("test")
+ .subscribe();
+
+ producer.sendAsync(value1.getBytes());
+ producer.sendAsync(value2.getBytes());
+ Transaction transaction = pulsarClient.newTransaction().withTransactionTimeout(1, TimeUnit.MINUTES)
+ .build().get();
+
+ Message<byte[]> message = consumer.receive();
+ assertEquals(value1, new String(message.getValue()));
+ // consumer receive the batch message one the first time, redeliverCount = 0
+ consumer.acknowledgeAsync(message.getMessageId(), transaction).get();
+
+ transaction.abort().get();
+
+ // consumer will receive the batch message two and then receive
+ // the message one and message two again, redeliverCount = 1
+ for (int i = 0; i < 3; i ++) {
+ message = consumer.receive();
+ }
+
+ transaction = pulsarClient.newTransaction().withTransactionTimeout(5, TimeUnit.MINUTES)
+ .build().get();
+
+ assertEquals(value2, new String(message.getValue()));
+ // consumer receive the batch message two the second time, redeliverCount = 1, also can be received
+ consumer.acknowledgeAsync(message.getMessageId(), transaction).get();
+
+ transaction.abort().get();
+
+ // consumer receive the batch message the third time, redeliverCount = 2,
+ // the message will be sent to DLQ, can't receive
+ assertNull(consumer.receive(3, TimeUnit.SECONDS));
+
+ assertEquals(value1, new String(deadLetterConsumer.receive(3, TimeUnit.SECONDS).getValue()));
+ assertEquals(value2, new String(deadLetterConsumer.receive(3, TimeUnit.SECONDS).getValue()));
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 8494e4804bb..f48ae2ed280 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1534,8 +1534,15 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
discardCorruptedMessage(messageId, cnx, ValidationError.BatchDeSerializeError);
}
- if (possibleToDeadLetter != null && possibleSendToDeadLetterTopicMessages != null) {
- possibleSendToDeadLetterTopicMessages.put(batchMessage, possibleToDeadLetter);
+ if (deadLetterPolicy != null && possibleSendToDeadLetterTopicMessages != null) {
+ if (redeliveryCount >= deadLetterPolicy.getMaxRedeliverCount()) {
+ possibleSendToDeadLetterTopicMessages.put(batchMessage,
+ possibleToDeadLetter);
+ if (redeliveryCount > deadLetterPolicy.getMaxRedeliverCount()) {
+ redeliverUnacknowledgedMessages(Collections.singleton(batchMessage));
+ return;
+ }
+ }
}
if (log.isDebugEnabled()) {