PIP-22: Dead Letter Topic #2508

codelipenghui · 2018-09-03T09:45:19Z

Motivation

Fixes #189

When consumer got messages from pulsar, It's difficult to ensure every message can be consume success. Pulsar support message redelivery feature by set acknowledge timeout when create a new consumer. This is a good feature guarantee consumer will not lost messages.

But however, some message will redelivery so many times possible, even to the extent that it can be never stop.

So, It's necessary to support a feature to control it by pulsar. Users can use this feature and customize this feature to control the message redelivery behavior. The feature named Dead Letter Topic.

Modifications

Consumer can set maximum number of redeliveries by java client.
Consumer can set the name of Dead Letter Topic by java client, It’s not necessary.
Message exceeding the maximum number of redeliveries should send to Dead Letter Topic and acknowledged automatic.

Result

If consumer enable future of dead letter topic. When Message exceeding the maximum number of redeliveries, message will send to the Dead Letter Topic and acknowledged automatic.

sijie · 2018-09-04T20:23:44Z

retest this please

sijie · 2018-09-04T20:26:08Z

@rdhabalia @merlimat can you help reviewing @codelipenghui 's pull request?

# Conflicts: # pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java

sijie · 2018-09-06T07:44:05Z

@rdhabalia @merlimat please review this. we need to include this for 2.2 release.

sijie · 2018-09-10T17:53:59Z

ping @rdhabalia @merlimat

merlimat

@codelipenghui Sorry for delay. The change mostly LGTM. Just left few comments.

merlimat · 2018-09-11T03:19:23Z

...main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentRedeliveryTracker.java

+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class NonPersistentRedeliveryTracker implements RedeliveryTracker {


Why is this tracker called "non-persistent"? Is it because the counter itself is not stored?

In that case it's a bit confusing since there's the concept of non-persistent topics as well, which I think is orthogonal to the redelivery tracker.

Because in the discuss of DLQ, has a solution is store the redelivery count in managed ledger, so i think it's possible to implement a PersistentRedeliveryTracker, so i named current implement to NonPersistentRedeliveryTracker.

Sure, so then maybe we can rename into InMemoryRedeliveryTracker to avoid confusion with this be associated with NonPersistentTopic?

merlimat · 2018-09-11T03:19:44Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -97,6 +100,7 @@ public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCurso
        this.name = topic.getName() + " / " + Codec.decode(cursor.getName());
        this.topic = topic;
        this.messagesToReplay = new ConcurrentLongPairSet(512, 2);
+        this.redeliveryTracker = new NonPersistentRedeliveryTracker();


See above, why NonPersistentRedeliveryTracker ?

merlimat · 2018-09-11T03:32:37Z

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DeadLetterPolicy.java

+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.policies.data;


I'd prefer to keep this class in pulsar-client package instead of pulsar-common, under the org.apache.pulsar.client.api package.

merlimat · 2018-09-11T03:32:42Z

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/DeadLetterPolicy.java

+        this.maxRedeliverCount = maxRedeliverCount;
+    }
+
+    public DeadLetterPolicy(int maxRedeliverCount, String deadLetterTopic) {


Instead of providing constructor overloads, we could use interface and builder approach like in other parts of the API. That way the code will long more descriptive.

Eg:

/// This DeadLetterPolicy.builder() .maxRedeliverCount(10) .deadLetterTopic("my-dql-topic") .build() // Instead of new DeadLetterPolicy(10, "my-dql-topic")

merlimat · 2018-09-11T03:33:09Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -1184,6 +1239,40 @@ public void redeliverUnacknowledgedMessages(Set<MessageId> messageIds) {
            MessageIdData.Builder builder = MessageIdData.newBuilder();
            batches.forEach(ids -> {
                List<MessageIdData> messageIdDatas = ids.stream().map(messageId -> {
+                    List<MessageImpl<T>> deadLetterMessages = null;


Can this new portion of code be factored out in a separate method that is called from here?

merlimat · 2018-09-11T03:33:18Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java

@@ -338,4 +340,9 @@
     * @return consumer builder.
     */
    ConsumerBuilder<T> intercept(ConsumerInterceptor<T> ...interceptors);
+
+    /**
+     * Set dead letter policy for consumer


Please expand the comment, explaining what are the purpose and scope of dead-letter policy (for people who are not familiar with the concept).
Also it would be nice to provide here a quick code example that explain how to enable this feature.

merlimat · 2018-09-11T03:37:28Z

pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java

        CommandMessage.Builder msgBuilder = CommandMessage.newBuilder();
        msgBuilder.setConsumerId(consumerId);
        msgBuilder.setMessageId(messageId);
+        msgBuilder.setRedeliveryCount(redeliveryCount);


if the redeliveryCount is 0, which is the default in the proto definition, and it's passed when the feature is disabled, we should avoid setting the value on the protobuf builder. If that is omitted, it would save 2 bytes per each message.

merlimat · 2018-09-11T03:41:50Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -265,7 +265,8 @@ public SendMessageInfo sendMessages(final List<Entry> entries, SendListener list
                if (i == (entries.size() - 1)) {
                    promise = writePromise;
                }
-                ctx.write(Commands.newMessage(consumerId, messageId, metadataAndPayload), promise);
+                int redeliveryCount = subscription.getDispatcher().getRedeliveryTracker().getRedeliveryCount(PositionImpl.get(messageId.getLedgerId(), messageId.getEntryId()));


Instead of keeping a separate redelivery tracker, would it be possible to reuse the pendingAcks maps on the Consumer class?

pendingAcks is a map for which the key is a pair longs and the value is a pair of longs as well.

(ledgerId, entryId) --> (batchSize, none)

As you can see, the second long field in the value is currently unused. We could keep the counter there.

After i check the pending acks in consumer, i recommend use redeliveryTracker to track the redelivery count:

First, in a subscription only have one redelivery tracker, pending acks is binding to consumer, we don't need to record redelivery count per consumer and if record it in consumer, will get a different behavior ismax redelivery count per consumer, but for now we defined the redelivery count per subscription.

Second, i think we need separate pending acks and redeliveryTracker, it's better to implement another redeliveryTracker.

First, in a subscription only have one redelivery tracker, pending acks is binding to consumer, we don't need to record redelivery count per consumer and if record it in consumer, will get a different behavior ismax redelivery count per consumer, but for now we defined the redelivery count per subscription.

Sure, what I was thinking was more to store it in consumer pending acks and carrying it over to the new consumer when redelivering.

Second, i think we need separate pending acks and redeliveryTracker, it's better to implement another redeliveryTracker.

Sure, these are separate concepts. Though I think the current name pendingAcks is a bit misleading. What the map really tracks is "message that were pushed to this consumer and not acked".

My main concern with the new map is to have to potentially accumulate lots of items in the concurrent hash map, irrespective of whether the DLQ is being used or not. Especially, this might be mostly not the case in exclusive/failover subscriptions, where data is typically processed in order.

My main concern with the new map is to have to potentially accumulate lots of items in the concurrent hash map,

I'm understand your concern, i think the tracker map is limited by maxUnackMessages, only put element into tracker map when trigger by redeliverUnackedMessages().

irrespective of whether the DLQ is being used or not.

I think we can optimize it next step, i think we can add a field in proto like enableRedeliveryTracker, broker just enable this mechanism when enableRedeliveryTracker is true.

Especially, this might be mostly not the case in exclusive/failover subscriptions, where data is typically processed in order.

Yes, i'm agree, i think we can only enable the counter tracking only on Shared subscriptions, mean that just in PersistentDispatcherMultipleConsumers.

codelipenghui · 2018-09-11T07:28:42Z

@merlimat I'm already address your comment except using pendingAcks to instead RedeliveryTracker.

merlimat · 2018-09-13T04:31:56Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -195,6 +195,7 @@ public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<St
                log.debug("[{}][{}] Individual acks on {}", topicName, subName, positions);
            }
            cursor.asyncDelete(positions, deleteCallback, positions);
+            dispatcher.getRedeliveryTracker().removeBatch(positions);


This will not clean up the redelivery tracker when doing a "cumulative-ack". It should be also cleaned at line 192, though for cumulative acks there might be a sequence of entries to clean-up and the concurrent hash map might not be a good fit (since we'd need sorted map).

Given that, and can we initially only enable the counter only on Shared subscriptions?

Given that, and can we initially only enable the counter only on Shared subscriptions?

I'm agree with your comment(Especially, this might be mostly not the case in exclusive/failover subscriptions, where data is typically processed in order).

This will not clean up the redelivery tracker when doing a "cumulative-ack". It should be also cleaned at line 192, though for cumulative acks there might be a sequence of entries to clean-up and the concurrent hash map might not be a good fit (since we'd need sorted map).

If we just enable the redelivery tracker in share mode, we don't need a sorted map, from client can't do cumulative-ack in share mode.

codelipenghui · 2018-09-13T06:44:11Z

@merlimat
To summarize, i need to disable the RedeliveryTracker in PersistentDispatcherSingleActiveConsumer.

Next step i will start a new PR to optimize the mechanism of only enable RedeliveryTracker by client subscription(enableRedeliveryTracker in subscription proto).

merlimat

👍

merlimat · 2018-09-13T11:01:50Z

retest this please

sijie · 2018-09-13T21:51:20Z

run java8 tests
run integration tests

merlimat · 2018-09-13T23:17:35Z

run java8 tests

sijie · 2018-09-14T05:38:35Z

run java8 tests
run integration tests

sijie · 2018-09-14T07:38:17Z

@codelipenghui it is a great contribution! thank you!

Reverts: #5881 ### Motivation The `redeliveryCount` was introduced in [PIP 22](https://github.com/apache/pulsar/wiki/PIP-22%3A-Pulsar-Dead-Letter-Topic) with this PR #2508. It is an extra field on a message that indicates how many times a message has been redelivered. In the original design, it was only incremented for shared subscriptions when the consumer sent `REDELIVER_UNACKNOWLEDGED_MESSAGES` to the broker. In #5881, this field's logic changed so that it is incremented each time a broker delivers a message to a consumer (after the initial delivery). The problem with this logic is that it counts messages that are sent to a consumer's `receiveQueue`, but not actually received by the client application, as "delivered" messages. This is especially problematic for the DLQ implementation because it relies on the counter to track deliveries, and this eager incrementing of the `redeliveryCount` could lead to fewer retries than an application would like. This PR returns the broker's behavior to the original state before #5881. Note that the DLQ logic is only triggered by messages that hit their ack timeout or are negatively acknowledged. This means that in some cases, a message could be delivered many times to a `receiveQueue` and once to the application and then sent to the DLQ. Given that our DLQ implementation has an intentional preference towards over delivery instead of under delivery, I think this logic should be fixed. One of the consequences of this PR is that the message filter logic for redelivering messages triggers this logic for incrementing `redeliveryCount`. See this code here: https://github.com/apache/pulsar/blob/b1a29b520d34d60e60160e3a7b9b0e26926063ee/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L198-L206 I'll need feedback from someone more familiar with message filtering to understand if this is a problematic change. If it is, I think we might need to revisit the logic in `filterEntriesForConsumer`. ### Modifications * Revert the relevant changes from #5895. I kept the test that was added in the PR and modified the assertion. * Fix test assertion ordering and modify expected value to align with new paradigm. ### Verifying this change This change includes modifications to tests as well as existing test coverage. ### Does this pull request potentially affect one of the following parts: This change is a break in current behavior, so I will send an email to the dev mailing list: https://lists.apache.org/thread/ts9d6zbtlz3y5xtv7p0c3dslk0vljpj2. ### Documentation - [x] `doc-not-needed`

Reverts: apache#5881 ### Motivation The `redeliveryCount` was introduced in [PIP 22](https://github.com/apache/pulsar/wiki/PIP-22%3A-Pulsar-Dead-Letter-Topic) with this PR apache#2508. It is an extra field on a message that indicates how many times a message has been redelivered. In the original design, it was only incremented for shared subscriptions when the consumer sent `REDELIVER_UNACKNOWLEDGED_MESSAGES` to the broker. In apache#5881, this field's logic changed so that it is incremented each time a broker delivers a message to a consumer (after the initial delivery). The problem with this logic is that it counts messages that are sent to a consumer's `receiveQueue`, but not actually received by the client application, as "delivered" messages. This is especially problematic for the DLQ implementation because it relies on the counter to track deliveries, and this eager incrementing of the `redeliveryCount` could lead to fewer retries than an application would like. This PR returns the broker's behavior to the original state before apache#5881. Note that the DLQ logic is only triggered by messages that hit their ack timeout or are negatively acknowledged. This means that in some cases, a message could be delivered many times to a `receiveQueue` and once to the application and then sent to the DLQ. Given that our DLQ implementation has an intentional preference towards over delivery instead of under delivery, I think this logic should be fixed. One of the consequences of this PR is that the message filter logic for redelivering messages triggers this logic for incrementing `redeliveryCount`. See this code here: https://github.com/apache/pulsar/blob/b1a29b520d34d60e60160e3a7b9b0e26926063ee/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L198-L206 I'll need feedback from someone more familiar with message filtering to understand if this is a problematic change. If it is, I think we might need to revisit the logic in `filterEntriesForConsumer`. ### Modifications * Revert the relevant changes from apache#5895. I kept the test that was added in the PR and modified the assertion. * Fix test assertion ordering and modify expected value to align with new paradigm. ### Verifying this change This change includes modifications to tests as well as existing test coverage. ### Does this pull request potentially affect one of the following parts: This change is a break in current behavior, so I will send an email to the dev mailing list: https://lists.apache.org/thread/ts9d6zbtlz3y5xtv7p0c3dslk0vljpj2. ### Documentation - [x] `doc-not-needed` (cherry picked from commit 2fd3509)

Reverts: apache#5881 ### Motivation The `redeliveryCount` was introduced in [PIP 22](https://github.com/apache/pulsar/wiki/PIP-22%3A-Pulsar-Dead-Letter-Topic) with this PR apache#2508. It is an extra field on a message that indicates how many times a message has been redelivered. In the original design, it was only incremented for shared subscriptions when the consumer sent `REDELIVER_UNACKNOWLEDGED_MESSAGES` to the broker. In apache#5881, this field's logic changed so that it is incremented each time a broker delivers a message to a consumer (after the initial delivery). The problem with this logic is that it counts messages that are sent to a consumer's `receiveQueue`, but not actually received by the client application, as "delivered" messages. This is especially problematic for the DLQ implementation because it relies on the counter to track deliveries, and this eager incrementing of the `redeliveryCount` could lead to fewer retries than an application would like. This PR returns the broker's behavior to the original state before apache#5881. Note that the DLQ logic is only triggered by messages that hit their ack timeout or are negatively acknowledged. This means that in some cases, a message could be delivered many times to a `receiveQueue` and once to the application and then sent to the DLQ. Given that our DLQ implementation has an intentional preference towards over delivery instead of under delivery, I think this logic should be fixed. One of the consequences of this PR is that the message filter logic for redelivering messages triggers this logic for incrementing `redeliveryCount`. See this code here: https://github.com/apache/pulsar/blob/b1a29b520d34d60e60160e3a7b9b0e26926063ee/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java#L198-L206 I'll need feedback from someone more familiar with message filtering to understand if this is a problematic change. If it is, I think we might need to revisit the logic in `filterEntriesForConsumer`. ### Modifications * Revert the relevant changes from apache#5895. I kept the test that was added in the PR and modified the assertion. * Fix test assertion ordering and modify expected value to align with new paradigm. ### Verifying this change This change includes modifications to tests as well as existing test coverage. ### Does this pull request potentially affect one of the following parts: This change is a break in current behavior, so I will send an email to the dev mailing list: https://lists.apache.org/thread/ts9d6zbtlz3y5xtv7p0c3dslk0vljpj2. ### Documentation - [x] `doc-not-needed`

codelipenghui added 9 commits August 30, 2018 15:18

Add redelivery_count to PulsarApi.proto.

9a61230

Implement Redelivery Tracker in broker.

7c0092a

Add DeadLetterPolicy.

41fb24a

Merge remote-tracking branch 'apache/master' into PIP-22

92a43a9

Implement dead letter topic in client.

dde79b7

Add UT to dead letter topic.

3c6a0c8

Fix bug in dead letter topic with multi topic subscription.

08a012d

Fix bug in send dead letter message then acknowledge.

793fc1f

Fix bugs in dead letter topic.

50ad8f7

codelipenghui mentioned this pull request Sep 4, 2018

PIP-22: Dead Letter Topic #2400

Closed

sijie requested review from merlimat, rdhabalia, hrsakai, jiazhai and sijie September 4, 2018 07:16

sijie assigned codelipenghui Sep 4, 2018

sijie added area/client type/feature The PR added a new feature or issue requested a new feature labels Sep 4, 2018

sijie added this to the 2.2.0-incubating milestone Sep 4, 2018

Do not instance dead letter producer in new consumerImpl().

b3a17e7

Merge remote-tracking branch 'apache/master' into PIP-22

e5f88cf

# Conflicts: # pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java

merlimat reviewed Sep 11, 2018

View reviewed changes

codelipenghui added 2 commits September 11, 2018 11:45

Merge remote-tracking branch 'apache/master' into PIP-22

5ba2168

optimization of dead letter topic.

903c4a0

merlimat reviewed Sep 13, 2018

View reviewed changes

Merge remote-tracking branch 'apache/master' into PIP-22

8bccab9

Disable RedeliveryTracker in PersistentDispatcherSingleActiveConsumer

df07259

merlimat approved these changes Sep 13, 2018

View reviewed changes

sijie approved these changes Sep 13, 2018

View reviewed changes

sijie merged commit 95fe84c into apache:master Sep 14, 2018

codelipenghui deleted the PIP-22 branch September 17, 2018 00:00

frejonb mentioned this pull request Nov 10, 2022

[C++ client] Deadletter queue functionality apache/pulsar-client-cpp#114

Closed

sijie mentioned this pull request Dec 27, 2019

ISSUE-5672: [C++ client] Deadletter queue functionality streamnative/pulsar-archived#38

Open

coderzc mentioned this pull request May 30, 2022

Message has no redelivery count if subscription type is Exclusive or Failover #15836

Open

michaeljmarshall mentioned this pull request Aug 11, 2022

[fix][broker] Revert 5895: fix redeliveryCount #17060

Merged

1 task

shibd mentioned this pull request Oct 27, 2022

[feature request] C++ and Python client support Dead Letter Topic apache/pulsar-client-cpp#77

Closed

2 tasks

sijie mentioned this pull request Aug 17, 2022

ISSUE-17139: [feature request] C++ and Python client support Dead Letter Topic streamnative/pulsar-archived#4748

Open

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PIP-22: Dead Letter Topic #2508

PIP-22: Dead Letter Topic #2508

codelipenghui commented Sep 3, 2018 •

edited

Loading

sijie commented Sep 4, 2018

sijie commented Sep 4, 2018

sijie commented Sep 6, 2018

sijie commented Sep 10, 2018

merlimat left a comment

merlimat Sep 11, 2018

codelipenghui Sep 11, 2018

merlimat Sep 11, 2018

codelipenghui Sep 11, 2018

merlimat Sep 11, 2018

merlimat Sep 11, 2018

codelipenghui Sep 11, 2018

merlimat Sep 11, 2018

codelipenghui Sep 11, 2018

merlimat Sep 11, 2018

codelipenghui Sep 11, 2018

merlimat Sep 11, 2018

codelipenghui Sep 11, 2018

merlimat Sep 11, 2018

merlimat Sep 11, 2018

codelipenghui Sep 11, 2018

merlimat Sep 13, 2018

codelipenghui Sep 13, 2018

codelipenghui commented Sep 11, 2018

merlimat Sep 13, 2018

codelipenghui Sep 13, 2018

codelipenghui commented Sep 13, 2018

merlimat left a comment

merlimat commented Sep 13, 2018

sijie commented Sep 13, 2018

merlimat commented Sep 13, 2018

sijie commented Sep 14, 2018

sijie commented Sep 14, 2018

PIP-22: Dead Letter Topic #2508

PIP-22: Dead Letter Topic #2508

Conversation

codelipenghui commented Sep 3, 2018 • edited Loading

Motivation

Modifications

Result

sijie commented Sep 4, 2018

sijie commented Sep 4, 2018

sijie commented Sep 6, 2018

sijie commented Sep 10, 2018

merlimat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Sep 11, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Sep 13, 2018

merlimat left a comment

Choose a reason for hiding this comment

merlimat commented Sep 13, 2018

sijie commented Sep 13, 2018

merlimat commented Sep 13, 2018

sijie commented Sep 14, 2018

sijie commented Sep 14, 2018

codelipenghui commented Sep 3, 2018 •

edited

Loading