init crCh for retry topic when start consumer

apache · Oct 14, 2022 · 003b83f · 003b83f
1 parent cbbe49c
commit 003b83f
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 0 deletions.
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
@@ -155,6 +155,7 @@ func (pc *pushConsumer) Start() error {
 			return
 		}
 
+		pc.handleRetryTopic()
 		go func() {
 			// todo start clean msg expired
 			for {
@@ -238,6 +239,13 @@ func (pc *pushConsumer) Start() error {
 	return err
 }
 
+func (pc *pushConsumer) handleRetryTopic() {
+	retryTopic := internal.GetRetryTopic(pc.consumerGroup)
+	if _, ok := pc.crCh[retryTopic]; !ok {
+		pc.crCh[retryTopic] = make(chan struct{}, pc.defaultConsumer.option.ConsumeGoroutineNums)
+	}
+}
+
 func (pc *pushConsumer) Shutdown() error {
 	var err error
 	pc.closeOnce.Do(func() {

diff --git a/internal/constants.go b/internal/constants.go
@@ -17,6 +17,8 @@ limitations under the License.
 
 package internal
 
+import "strings"
+
 const (
 	RetryGroupTopicPrefix    = "%RETRY%"
 	DefaultConsumerGroup     = "DEFAULT_CONSUMER"
@@ -31,5 +33,8 @@ func GetReplyTopic(clusterName string) string {
 }
 
 func GetRetryTopic(group string) string {
+	if strings.HasPrefix(group, RetryGroupTopicPrefix) {
+		return group
+	}
 	return RetryGroupTopicPrefix + group
 }