feat: consumer close log

zeromicro · May 24, 2024 · fb83318 · fb83318
1 parent 328f46c
commit fb83318
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/kq/queue.go b/kq/queue.go
@@ -162,6 +162,8 @@ func (q *kafkaQueue) Start() {
 	q.producerRoutines.Wait()
 	close(q.channel)
 	q.consumerRoutines.Wait()
+
+	logx.Infof("Consumer %s is closed", q.c.Name)
 }
 
 func (q *kafkaQueue) Stop() {
@@ -209,6 +211,7 @@ func (q *kafkaQueue) startProducers() {
 				// io.ErrClosedPipe means committing messages on the consumer,
 				// kafka will refire the messages on uncommitted messages, ignore
 				if err == io.EOF || err == io.ErrClosedPipe {
+					logx.Infof("Consumer %s-%d is closed, error: %q", q.c.Name, i, err.Error())
 					return
 				}
 				if err != nil {