[feat] PIP-468: V5 dead letter queue with scalable DLQ topic#25652
Merged
merlimat merged 2 commits intoapache:masterfrom May 2, 2026
Merged
[feat] PIP-468: V5 dead letter queue with scalable DLQ topic#25652merlimat merged 2 commits intoapache:masterfrom
merlimat merged 2 commits intoapache:masterfrom
Conversation
Move DLQ ownership from v4 per-segment ConsumerImpls up to the V5 ScalableQueueConsumer so that: - A single DLQ producer fans messages from every source segment, rather than one producer per segment. - The DLQ topic itself can be a scalable (topic://) topic. The v4 ConsumerImpl creates its DLQ producer via client.newProducer(...), which rejects the topic:// domain — V5 now uses its own producer builder, so a scalable DLQ "just works". - The default DLQ topic name (when the policy doesn't set one) is topic://<tenant>/<ns>/<source-local>-DLQ. ScalableTopicProducer's send path is now fully async: segment producers are stored as CompletableFuture so callers running on a netty IO thread (e.g. the V5 DLQ dispatch off a v4 receive callback) can chain instead of blocking on .get(). A per-segment dispatch chain serializes v4 sendAsync calls in user-call order; flushAsync awaits the snapshot of in-flight send futures. Tests cover explicit + default-named scalable DLQ targets, origin metadata preservation (REAL_TOPIC / REAL_SUBSCRIPTION / ORIGIN_MESSAGE_ID + key + properties + eventTime), and DLQ forwarding across multiple source segments.
merlimat
changed the title
# Conflicts: # pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/QueueConsumerBuilderV5.java # pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Move DLQ ownership from v4 per-segment
ConsumerImpls up to the V5ScalableQueueConsumerso that:topic://) topic. The v4ConsumerImplcreates its DLQ producer viaclient.newProducer(...), which rejects thetopic://domain — V5 now uses its own producer builder, so a scalable DLQ "just works".topic://<tenant>/<ns>/<source-local>-DLQ.ScalableTopicProducer's send path is now fully async: segment producers are stored asCompletableFutureso callers running on a netty IO thread (e.g. the V5 DLQ dispatch off a v4 receive callback) can chain instead of blocking on.get(). A per-segment dispatch chain serializes v4sendAsynccalls in user-call order;flushAsyncawaits the snapshot of in-flight send futures.Test plan
V5DeadLetterPolicyTest:testMessageGoesToScalableDlqWhenExplicitlyConfigured— explicit scalable DLQ target.testMessageGoesToDefaultScalableDlqTopic— default DLQ name (<source>-DLQ).testDlqMessagePreservesKeyPropertiesAndOriginMetadata— verifies key, user properties, eventTime, and origin metadata (REAL_TOPIC/REAL_SUBSCRIPTION/ORIGIN_MESSAGE_ID) are attached.testDlqAcrossMultipleSourceSegments— 3-segment source, 6 keys, single shared V5-side DLQ producer.org.apache.pulsar.client.api.v5.*) green: 113/113 pass.pulsar-client-v5andpulsar-brokercheckstyle clean.