[ISSUE #942] correct name typo (#943)

Co-authored-by: wuxb02 <[email protected]> Co-authored-by: dinglei <[email protected]>
apache · Oct 22, 2022 · ff80272 · ff80272
1 parent 50596e6
commit ff80272
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 6 deletions.
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
@@ -115,7 +115,7 @@ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error) {
 		consumeRequestCache: make(chan *ConsumeRequest, 4),
 	}
 	dc.mqChanged = c.messageQueueChanged
-	c.submitToConsume = c.consumeMessageCurrently
+	c.submitToConsume = c.consumeMessageConcurrently
 	c.interceptor = primitive.ChainInterceptors(c.option.Interceptors...)
 	return c, nil
 }
@@ -828,13 +828,13 @@ func (pc *defaultPullConsumer) correctTagsOffset(pr *PullRequest) {
 	}
 }
 
-func (pc *defaultPullConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.MessageQueue) {
+func (pc *defaultPullConsumer) consumeMessageConcurrently(pq *processQueue, mq *primitive.MessageQueue) {
 	msgList := pq.getMessages()
 	if msgList == nil {
 		return
 	}
 	if pq.IsDroppd() {
-		rlog.Info("defaultPullConsumer consumeMessageCurrently the message queue not be able to consume, because it was dropped", map[string]interface{}{
+		rlog.Info("defaultPullConsumer consumeMessageConcurrently the message queue not be able to consume, because it was dropped", map[string]interface{}{
 			rlog.LogKeyMessageQueue:  mq.String(),
 			rlog.LogKeyConsumerGroup: pc.consumerGroup,
 		})

diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
@@ -122,7 +122,7 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 	if p.consumeOrderly {
 		p.submitToConsume = p.consumeMessageOrderly
 	} else {
-		p.submitToConsume = p.consumeMessageCurrently
+		p.submitToConsume = p.consumeMessageConcurrently
 	}
 
 	p.interceptor = primitive.ChainInterceptors(p.option.Interceptors...)
@@ -1038,7 +1038,7 @@ func (pc *pushConsumer) resetRetryAndNamespace(subMsgs []*primitive.MessageExt)
 	}
 }
 
-func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.MessageQueue) {
+func (pc *pushConsumer) consumeMessageConcurrently(pq *processQueue, mq *primitive.MessageQueue) {
 	msgs := pq.getMessages()
 	if msgs == nil {
 		return
@@ -1072,7 +1072,7 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.
 		go primitive.WithRecover(func() {
 			defer func() {
 				if err := recover(); err != nil {
-					rlog.Error("consumeMessageCurrently panic", map[string]interface{}{
+					rlog.Error("consumeMessageConcurrently panic", map[string]interface{}{
 						rlog.LogKeyUnderlayError: err,
 						rlog.LogKeyConsumerGroup: pc.consumerGroup,
 					})