https://github.com/Shopify/sarama/issues/1192

sudoblockio · Oct 14, 2021 · f8bf90b · f8bf90b
1 parent f1089b8
commit f8bf90b
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/src/kafka/consumer.go b/src/kafka/consumer.go
@@ -167,6 +167,9 @@ func (c *ClaimConsumer) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sar
 		case <-time.After(5 * time.Second):
 			zap.S().Info("GROUP=", c.group, ",TOPIC=", c.topicName, " - No new kafka messages, waited 5 secs...")
 			continue
+		case <-sess.Context().Done():
+			zap.S().Warn("GROUP=", c.group, ",TOPIC=", c.topicName, " - Session is done, exiting ConsumeClaim loop...")
+			return nil
 		}
 
 		zap.S().Info("GROUP=", c.group, ",TOPIC=", c.topicName, ",PARTITION=", topicMsg.Partition, ",OFFSET=", topicMsg.Offset, " - New message")