You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ma...@apache.org on 2022/07/15 10:22:07 UTC
[pulsar] branch branch-2.9 updated: [improve][client] Add message key if exists to deadLetter messages (#16615)
This is an automated email from the ASF dual-hosted git repository.
mattisonchao pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new c121d0008bf [improve][client] Add message key if exists to deadLetter messages (#16615)
c121d0008bf is described below
commit c121d0008bf8befdb02ad1c40315e2b190da61d3
Author: gaozhangmin <ga...@qq.com>
AuthorDate: Fri Jul 15 18:03:29 2022 +0800
[improve][client] Add message key if exists to deadLetter messages (#16615)
(cherry picked from commit 8c528c4dc427c8456b3bdff8a35b273bba1d31c3)
---
.../pulsar/client/api/DeadLetterTopicTest.java | 60 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 10 ++--
2 files changed, 67 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
index 22b224990ed..9dfe3cf9aca 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
@@ -63,6 +63,66 @@ public class DeadLetterTopicTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ @Test
+ public void testDeadLetterTopicWithMessageKey() throws Exception {
+ final String topic = "persistent://my-property/my-ns/dead-letter-topic";
+
+ final int maxRedeliveryCount = 1;
+
+ final int sendMessages = 100;
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topic(topic)
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Shared)
+ .ackTimeout(1, TimeUnit.SECONDS)
+ .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).build())
+ .receiverQueueSize(100)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ @Cleanup
+ PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);// Creates new client connection
+ Consumer<byte[]> deadLetterConsumer = newPulsarClient.newConsumer(Schema.BYTES)
+ .topic("persistent://my-property/my-ns/dead-letter-topic-my-subscription-DLQ")
+ .subscriptionName("my-subscription")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(topic)
+ .create();
+
+ for (int i = 0; i < sendMessages; i++) {
+ producer.newMessage()
+ .key("test-key")
+ .value(String.format("Hello Pulsar [%d]", i).getBytes())
+ .send();
+ }
+
+ producer.close();
+
+ int totalReceived = 0;
+ do {
+ Message<byte[]> message = consumer.receive();
+ log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
+ totalReceived++;
+ } while (totalReceived < sendMessages * (maxRedeliveryCount + 1));
+
+ int totalInDeadLetter = 0;
+ do {
+ Message message = deadLetterConsumer.receive();
+ assertEquals(message.getKey(), "test-key");
+ log.info("dead letter consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
+ deadLetterConsumer.acknowledge(message);
+ totalInDeadLetter++;
+ } while (totalInDeadLetter < sendMessages);
+
+ deadLetterConsumer.close();
+ consumer.close();
+ }
+
+
@Test(groups = "quarantine")
public void testDeadLetterTopic() throws Exception {
final String topic = "persistent://my-property/my-ns/dead-letter-topic";
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 09f2685eb3c..46a6264ae73 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
@@ -1815,10 +1815,14 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
for (MessageImpl<T> message : finalDeadLetterMessages) {
String originMessageIdStr = getOriginMessageIdStr(message);
String originTopicNameStr = getOriginTopicNameStr(message);
- producerDLQ.newMessage(Schema.AUTO_PRODUCE_BYTES(message.getReaderSchema().get()))
+ TypedMessageBuilder<byte[]> typedMessageBuilderNew =
+ producerDLQ.newMessage(Schema.AUTO_PRODUCE_BYTES(message.getReaderSchema().get()))
.value(message.getData())
- .properties(getPropertiesMap(message, originMessageIdStr, originTopicNameStr))
- .sendAsync()
+ .properties(getPropertiesMap(message, originMessageIdStr, originTopicNameStr));
+ if (message.hasKey()) {
+ typedMessageBuilderNew.key(message.getKey());
+ }
+ typedMessageBuilderNew.sendAsync()
.thenAccept(messageIdInDLQ -> {
possibleSendToDeadLetterTopicMessages.remove(finalMessageId);
acknowledgeAsync(finalMessageId).whenComplete((v, ex) -> {