You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2023/12/08 09:53:46 UTC
(pulsar-client-go) branch master updated: [fix] fix channel deadlock in regexp consumer (#1141)
This is an automated email from the ASF dual-hosted git repository.
zike pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new 72aed95a [fix] fix channel deadlock in regexp consumer (#1141)
72aed95a is described below
commit 72aed95ae17baa4aa5ad8db7fb4abbf96f6c1c59
Author: Gonçalo Rodrigues <go...@gmail.com>
AuthorDate: Fri Dec 8 10:53:41 2023 +0100
[fix] fix channel deadlock in regexp consumer (#1141)
### Motivation
When using a regexp consumer, there's a race condition between producing and consuming new discovered topics. If the discover topic takes too long or the auto discovery period is too short, then multiple ticker.C messages may be processed in a row which will block on the subcsribe/unsubscribe channels as they only have a buffer size of 1. This will block new topics from being discovered forever.
### Modifications
Moved the consumers into their own goroutines and use an unbuffered channel. There's multiple ways to go about it but it's good practice to keep consumers and producers separate. Consumers are run until the channels they are consumed from are closed, which happens when the producer (monitor) returns.
---
pulsar/consumer_regex.go | 31 ++++++++-------------
pulsar/consumer_regex_test.go | 63 +++++++++++++++++++++++++++++++++++++++++++
2 files changed, 74 insertions(+), 20 deletions(-)
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index 79e2293b..d36694ef 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -50,8 +50,6 @@ type regexConsumer struct {
consumersLock sync.Mutex
consumers map[string]Consumer
- subscribeCh chan []string
- unsubscribeCh chan []string
closeOnce sync.Once
closeCh chan struct{}
@@ -75,9 +73,7 @@ func newRegexConsumer(c *client, opts ConsumerOptions, tn *internal.TopicName, p
namespace: tn.Namespace,
pattern: pattern,
- consumers: make(map[string]Consumer),
- subscribeCh: make(chan []string, 1),
- unsubscribeCh: make(chan []string, 1),
+ consumers: make(map[string]Consumer),
closeCh: make(chan struct{}),
@@ -163,12 +159,11 @@ func (c *regexConsumer) Ack(msg Message) error {
return c.AckID(msg.ID())
}
-func (c *regexConsumer) ReconsumeLater(msg Message, delay time.Duration) {
+func (c *regexConsumer) ReconsumeLater(_ Message, _ time.Duration) {
c.log.Warnf("regexp consumer not support ReconsumeLater yet.")
}
-func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customProperties map[string]string,
- delay time.Duration) {
+func (c *regexConsumer) ReconsumeLaterWithCustomProperties(_ Message, _ map[string]string, _ time.Duration) {
c.log.Warnf("regexp consumer not support ReconsumeLaterWithCustomProperties yet.")
}
@@ -297,11 +292,11 @@ func (c *regexConsumer) Close() {
})
}
-func (c *regexConsumer) Seek(msgID MessageID) error {
+func (c *regexConsumer) Seek(_ MessageID) error {
return newError(SeekFailed, "seek command not allowed for regex consumer")
}
-func (c *regexConsumer) SeekByTime(time time.Time) error {
+func (c *regexConsumer) SeekByTime(_ time.Time) error {
return newError(SeekFailed, "seek command not allowed for regex consumer")
}
@@ -329,14 +324,6 @@ func (c *regexConsumer) monitor() {
if !c.closed() {
c.discover()
}
- case topics := <-c.subscribeCh:
- if len(topics) > 0 && !c.closed() {
- c.subscribe(topics, c.dlq, c.rlq)
- }
- case topics := <-c.unsubscribeCh:
- if len(topics) > 0 && !c.closed() {
- c.unsubscribe(topics)
- }
}
}
}
@@ -358,8 +345,12 @@ func (c *regexConsumer) discover() {
}).
Debug("discover topics")
- c.unsubscribeCh <- staleTopics
- c.subscribeCh <- newTopics
+ if len(staleTopics) > 0 {
+ c.unsubscribe(staleTopics)
+ }
+ if len(newTopics) > 0 {
+ c.subscribe(newTopics, c.dlq, c.rlq)
+ }
}
func (c *regexConsumer) knownTopics() []string {
diff --git a/pulsar/consumer_regex_test.go b/pulsar/consumer_regex_test.go
index de946134..3e5f1d61 100644
--- a/pulsar/consumer_regex_test.go
+++ b/pulsar/consumer_regex_test.go
@@ -241,6 +241,7 @@ func runRegexConsumerDiscoverPatternFoo(t *testing.T, c Client, namespace string
func TestRegexConsumer(t *testing.T) {
t.Run("MatchOneTopic", runWithClientNamespace(runRegexConsumerMatchOneTopic))
t.Run("AddTopic", runWithClientNamespace(runRegexConsumerAddMatchingTopic))
+ t.Run("AutoDiscoverTopics", runWithClientNamespace(runRegexConsumerAutoDiscoverTopics))
}
func runRegexConsumerMatchOneTopic(t *testing.T, c Client, namespace string) {
@@ -346,6 +347,68 @@ func runRegexConsumerAddMatchingTopic(t *testing.T, c Client, namespace string)
}
}
+func runRegexConsumerAutoDiscoverTopics(t *testing.T, c Client, namespace string) {
+ topicsPattern := fmt.Sprintf("persistent://%s/foo.*", namespace)
+ opts := ConsumerOptions{
+ TopicsPattern: topicsPattern,
+ SubscriptionName: "regex-sub",
+ // this is purposefully short to test parallelism between discover and subscribe calls
+ AutoDiscoveryPeriod: 1 * time.Nanosecond,
+ }
+ consumer, err := c.Subscribe(opts)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer consumer.Close()
+
+ topicInRegex1 := namespace + "/foo-topic-1"
+ p1, err := c.CreateProducer(ProducerOptions{
+ Topic: topicInRegex1,
+ DisableBatching: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer p1.Close()
+
+ topicInRegex2 := namespace + "/foo-topic-2"
+ p2, err := c.CreateProducer(ProducerOptions{
+ Topic: topicInRegex2,
+ DisableBatching: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer p2.Close()
+
+ time.Sleep(100 * time.Millisecond)
+
+ err = genMessages(p1, 5, func(idx int) string {
+ return fmt.Sprintf("foo-message-%d", idx)
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = genMessages(p2, 5, func(idx int) string {
+ return fmt.Sprintf("foo-message-%d", idx)
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ ctx := context.Background()
+ for i := 0; i < 10; i++ {
+ m, err := consumer.Receive(ctx)
+ if err != nil {
+ t.Errorf("failed to receive message error: %+v", err)
+ } else {
+ assert.Truef(t, strings.HasPrefix(string(m.Payload()), "foo-"),
+ "message does not start with foo: %s", string(m.Payload()))
+ }
+ }
+}
+
func genMessages(p Producer, num int, msgFn func(idx int) string) error {
ctx := context.Background()
for i := 0; i < num; i++ {