You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "geniusjoe (via GitHub)" <gi...@apache.org> on 2025/07/15 07:54:31 UTC
[PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
geniusjoe opened a new pull request, #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392
Fixes https://github.com/apache/pulsar-client-go/issues/1388
Master Issue: https://github.com/apache/pulsar-client-go/pull/907, https://github.com/apache/pulsar/pull/23182
### Motivation
Since pulsar sdk does not support `AckTimeout()` method, we can only use `Nack()` or `ReconsumeLater()` to trigger dlq policy and send messages to dlq. We can find system properties(e.g. `REAL_TOPIC` and `ORIGIN_MESSAGE_ID`) in these dlq messages.
Refer to issue https://github.com/apache/pulsar-client-go/issues/1388, before pr https://github.com/apache/pulsar-client-go/pull/907 `dlq_router#run()` will directly copy properties to create a new dlq message when receiving a message from `dlq.Chan()`. But we only define message properties in ReconsumeLater's `ReconsumeLaterWithCustomProperties()` and not in Nack's `dispatcher()` before send to this `dlq.Chan()`, which lead to only dlq messages created by `ReconsumeLater()` can have these properties.
The above pr replaced this behavior with just adding system properties as soon as creating dlq messages in `dlq_router#run()`, which use `message.topic` as `REAL_TOPIC` property value. But messages sent from `ReconsumeLater() ReconsumeLaterWithCustomProperties()` do not contain related `message.topic` fields, in case it will override defined `REAL_TOPIC` to empty string.
### Modifications
- Add `REAL_TOPIC`, `ORIGIN_MESSAGE_IDY_TIME`, `ORIGIN_MESSAGE_ID` system properties both in ReconsumeLater's `ReconsumeLaterWithCustomProperties()` and Nack's `dispatcher()` functions.
- Add `Key`, `OrderingKey`, `EventTime` message fields both in `ReconsumeLater() ReconsumeLaterWithCustomProperties()` and `dlq_router run()` functions to keep consistent with [Java realization](https://github.com/apache/pulsar/pull/23182), so that dlq messages can preserve source message info.
- Remove adding system properties strategy in `dlq_router#run()`.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change updated tests and can be verified as follows:
DLQWithProducerOptions
TestRLQ
### Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API: (no)
- The schema: (no)
- The default values of configurations: (no)
- The wire protocol: (no)
### Documentation
- Does this pull request introduce a new feature? (no)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392#discussion_r2257577576
##########
pulsar/consumer_test.go:
##########
@@ -1623,10 +1626,18 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
assert.True(t, regex.MatchString(msg.ProducerName()))
// check original messageId
+ assert.NotEmpty(t, msg.Properties()[SysPropertyOriginMessageID])
assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])
// check original topic
- assert.NotEmpty(t, msg.Properties()[SysPropertyRealTopic])
+ assert.Contains(t, msg.Properties()[SysPropertyRealTopic], topic)
+
+ // check original key
+ assert.NotEmpty(t, msg.Key())
+ assert.NotEmpty(t, msg.OrderingKey())
+
+ // check original event time
+ assert.NotEqual(t, 0, msg.EventTime())
Review Comment:
Thanks for your reply. I checked these values.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392#discussion_r2257577576
##########
pulsar/consumer_test.go:
##########
@@ -1623,10 +1626,18 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
assert.True(t, regex.MatchString(msg.ProducerName()))
// check original messageId
+ assert.NotEmpty(t, msg.Properties()[SysPropertyOriginMessageID])
assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])
// check original topic
- assert.NotEmpty(t, msg.Properties()[SysPropertyRealTopic])
+ assert.Contains(t, msg.Properties()[SysPropertyRealTopic], topic)
+
+ // check original key
+ assert.NotEmpty(t, msg.Key())
+ assert.NotEmpty(t, msg.OrderingKey())
+
+ // check original event time
+ assert.NotEqual(t, 0, msg.EventTime())
Review Comment:
Thanks for your reply. I checked these values in test cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392#discussion_r2235656562
##########
pulsar/consumer_test.go:
##########
@@ -1623,10 +1626,18 @@ func DLQWithProducerOptions(t *testing.T, prodOpt *ProducerOptions) {
assert.True(t, regex.MatchString(msg.ProducerName()))
// check original messageId
+ assert.NotEmpty(t, msg.Properties()[SysPropertyOriginMessageID])
assert.NotEmpty(t, msg.Properties()[PropertyOriginMessageID])
// check original topic
- assert.NotEmpty(t, msg.Properties()[SysPropertyRealTopic])
+ assert.Contains(t, msg.Properties()[SysPropertyRealTopic], topic)
+
+ // check original key
+ assert.NotEmpty(t, msg.Key())
+ assert.NotEmpty(t, msg.OrderingKey())
+
+ // check original event time
+ assert.NotEqual(t, 0, msg.EventTime())
Review Comment:
Could you check the actual value of `Key`, `OrderingKey`, `EventTime` and `SysPropertyOriginMessageID`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392#discussion_r2257011912
##########
pulsar/consumer_partition.go:
##########
@@ -1603,6 +1603,26 @@ func (pc *partitionConsumer) dispatcher() {
if !pc.isSeeking.Load() {
if pc.dlq.shouldSendToDlq(&nextMessage) {
// pass the message to the DLQ router
+ // we need to create a new ConsumerMessage and add dlq related metadata properties
Review Comment:
I do not find typo word "orinal", maybe I should ignore this suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie merged PR #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
Posted by "geniusjoe (via GitHub)" <gi...@apache.org>.
geniusjoe commented on code in PR #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392#discussion_r2257028453
##########
pulsar/consumer_partition.go:
##########
@@ -1603,6 +1603,26 @@ func (pc *partitionConsumer) dispatcher() {
if !pc.isSeeking.Load() {
if pc.dlq.shouldSendToDlq(&nextMessage) {
// pass the message to the DLQ router
+ // we need to create a new ConsumerMessage and add dlq related metadata properties
+ properties := make(map[string]string)
+ properties[SysPropertyRealTopic] = messages[0].Topic()
+ properties[SysPropertyOriginMessageID] = messages[0].msgID.String()
+ properties[PropertyOriginMessageID] = messages[0].msgID.String()
+ for key, value := range messages[0].properties {
+ properties[key] = value
+ }
+ nextMessage = ConsumerMessage{
+ Consumer: pc.parentConsumer,
+ // Copy msgID so that dlq/rlq router can ack this msg after successfully sent to new topic
+ Message: &message{
+ payLoad: messages[0].Payload(),
+ key: messages[0].Key(),
+ orderingKey: messages[0].OrderingKey(),
+ properties: properties,
+ eventTime: messages[0].EventTime(),
+ msgID: messages[0].msgID,
+ },
+ }
Review Comment:
Duplicated code only contain 6 lines, maybe we create a new helper function later to prevent overdesigning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Fix] DLQ Messages will lose system properties when sent from reconsumeLater() [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1392:
URL: https://github.com/apache/pulsar-client-go/pull/1392#discussion_r2212272483
##########
pulsar/consumer_partition.go:
##########
@@ -1603,6 +1603,26 @@ func (pc *partitionConsumer) dispatcher() {
if !pc.isSeeking.Load() {
if pc.dlq.shouldSendToDlq(&nextMessage) {
// pass the message to the DLQ router
+ // we need to create a new ConsumerMessage and add dlq related metadata properties
Review Comment:
There is a typo in the comment: 'orinal' should be 'original' in the context. The word should be 'dlq related metadata properties'.
##########
pulsar/consumer_partition.go:
##########
@@ -1603,6 +1603,26 @@ func (pc *partitionConsumer) dispatcher() {
if !pc.isSeeking.Load() {
if pc.dlq.shouldSendToDlq(&nextMessage) {
// pass the message to the DLQ router
+ // we need to create a new ConsumerMessage and add dlq related metadata properties
+ properties := make(map[string]string)
+ properties[SysPropertyRealTopic] = messages[0].Topic()
+ properties[SysPropertyOriginMessageID] = messages[0].msgID.String()
+ properties[PropertyOriginMessageID] = messages[0].msgID.String()
+ for key, value := range messages[0].properties {
+ properties[key] = value
+ }
+ nextMessage = ConsumerMessage{
+ Consumer: pc.parentConsumer,
+ // Copy msgID so that dlq/rlq router can ack this msg after successfully sent to new topic
+ Message: &message{
+ payLoad: messages[0].Payload(),
+ key: messages[0].Key(),
+ orderingKey: messages[0].OrderingKey(),
+ properties: properties,
+ eventTime: messages[0].EventTime(),
+ msgID: messages[0].msgID,
+ },
+ }
Review Comment:
[nitpick] The message construction logic is duplicated between this location and ReconsumeLaterWithCustomProperties. Consider extracting this into a helper function to reduce code duplication and improve maintainability.
```suggestion
nextMessage = createConsumerMessageWithProperties(
pc.parentConsumer,
messages[0],
properties,
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org