[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.

Author: goncalo-rodrigues

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 {

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++ {