fix(kafka): ConsumeClaim should return when messages channel closed (#2120)

Signed-off-by: cong <zhangcong1992@gmail.com> Signed-off-by: cong <zhangcong1992@gmail.com>
2022-09-24 00:54:59 +08:00 · 2022-09-24 00:54:59 +08:00 · 1a7882fca5
parent ac87ed4e9d
commit 1a7882fca5
1 changed files with 5 additions and 1 deletions
--- a/internal/component/kafka/consumer.go
+++ b/internal/component/kafka/consumer.go
@ -38,7 +38,11 @@ func (consumer *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai

 	for {
 		select {
-		case message := <-claim.Messages():
+		case message, ok := <-claim.Messages():
+			if !ok {
+				return nil
+			}
+
 			if consumer.k.consumeRetryEnabled {
 				if err := retry.NotifyRecover(func() error {
 					return consumer.doCallback(session, message)