update

pingcap · Dec 3, 2024 · 2da8c87 · 2da8c87
1 parent 58d19a1
commit 2da8c87
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/cmd/kafka-consumer/consumer.go b/cmd/kafka-consumer/consumer.go
@@ -125,6 +125,12 @@ func (c *consumer) Consume(ctx context.Context) {
 		}
 	}()
 	for {
+		select {
+		case <-ctx.Done():
+			log.Info("consumer exist: context cancelled")
+			break
+		default:
+		}
 		msg, err := c.client.ReadMessage(-1)
 		if err != nil {
 			log.Error("read message failed, just continue to retry", zap.Error(err))