You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "jordanfitz (via GitHub)" <gi...@apache.org> on 2026/01/08 18:11:28 UTC
[PR] Implement ReconsumeLater on regex consumer [pulsar-client-go]
jordanfitz opened a new pull request, #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456
### Motivation
Currently, `ReconsumeLater` and `ReconsumeLaterWithCustomProperties` are unimplemented on `regexConsumer`. From what I can tell, these appear to just be missing methods rather than some fundamental limitation of the consumer implementation. Hopefully I'm correct there.
### Modifications
I've essentially copied over the `ReconsumeLaterWithCustomProperties` implementation from `multiTopicConsumer`. I ran into the problem that the current code to add the RLQ topic to the consumer's `Topics` slice doesn't take into account subscriptions defined by `TopicsPattern`. To address this, I (perhaps naively) append the RLQ topic to the slice returned by `regexConsumer.topics()`.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- Added a test to ensure that a call to `ReconsumeLater` on a regex-based consumer causes the message to be published to the RLQ.
### 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
- If yes, how is the feature documented? N/A
- If a feature is not applicable for documentation, explain why? The behavior of the newly-implemented method should be the same as the existing consumer implementations.
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2689351938
##########
pulsar/consumer_regex.go:
##########
@@ -454,6 +476,11 @@ func (c *regexConsumer) topics() ([]string, error) {
}
filtered := filterTopics(topics, c.pattern)
+
+ if c.options.RetryEnable {
Review Comment:
Consider adding a nil check for c.options.DLQ before accessing c.options.DLQ.RetryLetterTopic for defensive programming. While the current code path through newConsumer ensures DLQ is initialized when RetryEnable is true, adding this check would make the code more robust and easier to maintain.
```suggestion
if c.options.RetryEnable && c.options.DLQ != nil {
```
##########
pulsar/consumer_regex.go:
##########
@@ -186,12 +186,34 @@ func (c *regexConsumer) Ack(msg Message) error {
return c.AckID(msg.ID())
}
-func (c *regexConsumer) ReconsumeLater(_ Message, _ time.Duration) {
- c.log.Warnf("regexp consumer not support ReconsumeLater yet.")
+func (c *regexConsumer) ReconsumeLater(msg Message, delay time.Duration) {
+ c.ReconsumeLaterWithCustomProperties(msg, map[string]string{}, delay)
}
-func (c *regexConsumer) ReconsumeLaterWithCustomProperties(_ Message, _ map[string]string, _ time.Duration) {
- c.log.Warnf("regexp consumer not support ReconsumeLaterWithCustomProperties yet.")
+func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string,
+ delay time.Duration) {
+ names, err := validateTopicNames(msg.Topic())
+ if err != nil {
+ c.log.Errorf("validate msg topic %q failed: %v", msg.Topic(), err)
+ return
+ }
+ if len(names) != 1 {
+ c.log.Errorf("invalid msg topic %q names: %+v ", msg.Topic(), names)
+ return
+ }
+
+ tn := names[0]
+ fqdnTopic := internal.TopicNameWithoutPartitionPart(tn)
+ consumer, ok := c.consumers[fqdnTopic]
+ if !ok {
+ // check to see if the topic with the partition part is in the consumers
+ // this can happen when the consumer is configured to consume from a specific partition
+ if consumer, ok = c.consumers[tn.Name]; !ok {
+ c.log.Warnf("consumer of topic %s not exist unexpectedly", msg.Topic())
+ return
+ }
+ }
+ consumer.ReconsumeLaterWithCustomProperties(msg, customProperties, delay)
Review Comment:
There's a potential race condition here. The consumers map is being accessed without holding consumersLock, while the discover() method (which runs in a separate goroutine) modifies this map with the lock held. This could lead to concurrent map read/write panics. The map access should be protected with consumersLock to ensure thread safety.
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "jordanfitz (via GitHub)" <gi...@apache.org>.
jordanfitz commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2695347866
##########
pulsar/consumer_regex.go:
##########
@@ -454,6 +481,11 @@ func (c *regexConsumer) topics() ([]string, error) {
}
filtered := filterTopics(topics, c.pattern)
+
+ if c.options.RetryEnable && c.options.DLQ != nil {
+ filtered = append(filtered, c.options.DLQ.RetryLetterTopic)
Review Comment:
I think I'm not entirely clear on how this issue could arise, but probably better to be defensive regardless 😄 Pushed up a change which checks the filtered topics for the RLQ topic as suggested.
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "jordanfitz (via GitHub)" <gi...@apache.org>.
jordanfitz commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2695345207
##########
pulsar/consumer_regex.go:
##########
@@ -186,12 +186,39 @@ func (c *regexConsumer) Ack(msg Message) error {
return c.AckID(msg.ID())
}
-func (c *regexConsumer) ReconsumeLater(_ Message, _ time.Duration) {
- c.log.Warnf("regexp consumer not support ReconsumeLater yet.")
+func (c *regexConsumer) ReconsumeLater(msg Message, delay time.Duration) {
+ c.ReconsumeLaterWithCustomProperties(msg, map[string]string{}, delay)
}
-func (c *regexConsumer) ReconsumeLaterWithCustomProperties(_ Message, _ map[string]string, _ time.Duration) {
- c.log.Warnf("regexp consumer not support ReconsumeLaterWithCustomProperties yet.")
+func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string,
+ delay time.Duration) {
+ names, err := validateTopicNames(msg.Topic())
+ if err != nil {
+ c.log.Errorf("validate msg topic %q failed: %v", msg.Topic(), err)
+ return
+ }
+ if len(names) != 1 {
+ c.log.Errorf("invalid msg topic %q names: %+v ", msg.Topic(), names)
+ return
+ }
+
+ tn := names[0]
+ fqdnTopic := internal.TopicNameWithoutPartitionPart(tn)
+
+ c.consumersLock.Lock()
+ defer c.consumersLock.Unlock()
+
+ consumer, ok := c.consumers[fqdnTopic]
+ if !ok {
+ // check to see if the topic with the partition part is in the consumers
+ // this can happen when the consumer is configured to consume from a specific partition
+ if consumer, ok = c.consumers[tn.Name]; !ok {
+ c.log.Warnf("consumer of topic %s not exist unexpectedly", msg.Topic())
+ return
+ }
+ }
+
Review Comment:
I went ahead and added a shared func called `reconsumeLaterWithMultipleTopics`
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2693169133
##########
pulsar/consumer_regex.go:
##########
@@ -454,6 +481,11 @@ func (c *regexConsumer) topics() ([]string, error) {
}
filtered := filterTopics(topics, c.pattern)
+
+ if c.options.RetryEnable && c.options.DLQ != nil {
+ filtered = append(filtered, c.options.DLQ.RetryLetterTopic)
Review Comment:
This make sense to me, PTAL @jordanfitz
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2693268450
##########
pulsar/consumer_regex.go:
##########
@@ -186,12 +186,39 @@ func (c *regexConsumer) Ack(msg Message) error {
return c.AckID(msg.ID())
}
-func (c *regexConsumer) ReconsumeLater(_ Message, _ time.Duration) {
- c.log.Warnf("regexp consumer not support ReconsumeLater yet.")
+func (c *regexConsumer) ReconsumeLater(msg Message, delay time.Duration) {
+ c.ReconsumeLaterWithCustomProperties(msg, map[string]string{}, delay)
}
-func (c *regexConsumer) ReconsumeLaterWithCustomProperties(_ Message, _ map[string]string, _ time.Duration) {
- c.log.Warnf("regexp consumer not support ReconsumeLaterWithCustomProperties yet.")
+func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string,
+ delay time.Duration) {
+ names, err := validateTopicNames(msg.Topic())
+ if err != nil {
+ c.log.Errorf("validate msg topic %q failed: %v", msg.Topic(), err)
+ return
+ }
+ if len(names) != 1 {
+ c.log.Errorf("invalid msg topic %q names: %+v ", msg.Topic(), names)
+ return
+ }
+
+ tn := names[0]
+ fqdnTopic := internal.TopicNameWithoutPartitionPart(tn)
+
+ c.consumersLock.Lock()
+ defer c.consumersLock.Unlock()
+
+ consumer, ok := c.consumers[fqdnTopic]
+ if !ok {
+ // check to see if the topic with the partition part is in the consumers
+ // this can happen when the consumer is configured to consume from a specific partition
+ if consumer, ok = c.consumers[tn.Name]; !ok {
+ c.log.Warnf("consumer of topic %s not exist unexpectedly", msg.Topic())
+ return
+ }
+ }
+
Review Comment:
https://github.com/apache/pulsar-client-go/blob/ffdc3afec23679e86b41ebb96b25c723f2ec5e05/pulsar/consumer_multitopic.go#L260-L283
This part of the code is basically the same as `multiTopicConsumer`. Can we reuse the related code?
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2696629964
##########
pulsar/consumer_regex.go:
##########
@@ -454,6 +481,11 @@ func (c *regexConsumer) topics() ([]string, error) {
}
filtered := filterTopics(topics, c.pattern)
+
+ if c.options.RetryEnable && c.options.DLQ != nil {
+ filtered = append(filtered, c.options.DLQ.RetryLetterTopic)
Review Comment:
> Appending the retry letter topic to the filtered topics list may cause issues during auto-discovery.
https://github.com/apache/pulsar-client-go/blob/918d9731363861acd967ae4460b48cf711f2bdbc/pulsar/consumer_regex.go#L397-L402
The `discover` function is called repeatedly; without write validation, `retryTopic` will be written repeatedly.
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2696638788
##########
pulsar/consumer_regex_test.go:
##########
@@ -513,3 +513,70 @@ func TestRegexTopicGetLastMessageIDs(t *testing.T) {
assert.Equal(t, 1, len(messages))
}
}
+
+func TestRegexConsumerReconsumeLater(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topic1 := fmt.Sprintf("regex-reconsume-topic-%v", time.Now().Nanosecond())
+ assert.Nil(t, createPartitionedTopic(topic1, 1))
+
+ topic2 := fmt.Sprintf("regex-reconsume-topic-%v", time.Now().Nanosecond())
Review Comment:
Using `time.Now().Nanosecond()` for generating unique topic names is unreliable because `Nanosecond()` only returns the nanosecond offset within the current second (0-999999999), not a unique timestamp. If the test runs multiple times within the same nanosecond offset of different seconds, or if topics are created in quick succession, name collisions can occur. Use `time.Now().UnixNano()` instead to generate truly unique topic names.
```suggestion
topic1 := fmt.Sprintf("regex-reconsume-topic-%v", time.Now().UnixNano())
assert.Nil(t, createPartitionedTopic(topic1, 1))
topic2 := fmt.Sprintf("regex-reconsume-topic-%v", time.Now().UnixNano())
```
##########
pulsar/consumer_regex.go:
##########
@@ -454,6 +458,14 @@ func (c *regexConsumer) topics() ([]string, error) {
}
filtered := filterTopics(topics, c.pattern)
+
+ if c.options.RetryEnable && c.options.DLQ != nil {
+ retryTopic := c.options.DLQ.RetryLetterTopic
+ if !slices.Contains(filtered, retryTopic) {
Review Comment:
The check `c.options.DLQ != nil` is insufficient. If `RetryEnable` is true but `RetryLetterTopic` is empty (which can happen if Comment 1's issue causes initialization to fail), this code will add an empty string to the filtered topics list, causing subscription errors. Add a check `retryTopic != \"\"` before appending to the filtered list.
```suggestion
if retryTopic != "" && !slices.Contains(filtered, retryTopic) {
```
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "jordanfitz (via GitHub)" <gi...@apache.org>.
jordanfitz commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2699162908
##########
pulsar/consumer_regex.go:
##########
@@ -454,6 +481,11 @@ func (c *regexConsumer) topics() ([]string, error) {
}
filtered := filterTopics(topics, c.pattern)
+
+ if c.options.RetryEnable && c.options.DLQ != nil {
+ filtered = append(filtered, c.options.DLQ.RetryLetterTopic)
Review Comment:
Gotcha. I think my confusion stems from my impression that the logic in `topics()` is each time producing a new slice of topics in the namespace (via its call to `GetTopicsOfNamespace`), to which the RLQ topic is appended afterward. So two calls to `topics()` _should_ result in two different slices, each with one RLQ topic, unless I'm mistaken. Maybe it's possible for the regex to match the RLQ topic, in which case, it would end up in the slice more than once. I think the defensive check is definitely a good option! 😄
--
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] Implement ReconsumeLater on regex consumer [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1456:
URL: https://github.com/apache/pulsar-client-go/pull/1456#discussion_r2691736738
##########
pulsar/consumer_regex.go:
##########
@@ -454,6 +481,11 @@ func (c *regexConsumer) topics() ([]string, error) {
}
filtered := filterTopics(topics, c.pattern)
+
+ if c.options.RetryEnable && c.options.DLQ != nil {
+ filtered = append(filtered, c.options.DLQ.RetryLetterTopic)
Review Comment:
Appending the retry letter topic to the filtered topics list may cause issues during auto-discovery. The `topics()` method is called periodically by the monitor goroutine to discover new topics matching the pattern. If the RLQ topic doesn't match the regex pattern, repeatedly appending it could lead to duplicate subscription attempts or incorrect behavior. Consider checking if the RLQ topic is already in the filtered list before appending, or ensure the RLQ topic is subscribed to separately during consumer initialization rather than through the discovery mechanism.
```suggestion
retryTopic := c.options.DLQ.RetryLetterTopic
alreadyPresent := false
for _, t := range filtered {
if t == retryTopic {
alreadyPresent = true
break
}
}
if !alreadyPresent {
filtered = append(filtered, retryTopic)
}
```
--
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