fix(kafka): should return when session context is done in ConsumeClaim
@@ -35,7 +35,10 @@ type consumer struct {
func (consumer *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
b := consumer.k.backOffConfig.NewBackOffWithContext(session.Context())
- for message := range claim.Messages() {
+
+ for {
+ select {
+ case message := <-claim.Messages():
if consumer.k.consumeRetryEnabled {
if err := retry.NotifyRecover(func() error {
return consumer.doCallback(session, message)
@@ -52,10 +55,14 @@ func (consumer *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
consumer.k.logger.Errorf("Error processing Kafka message: %s/%d/%d [key=%s]. Error: %v.", message.Topic, message.Partition, message.Offset, asBase64String(message.Key), err)
}
}
- }
-
+ // Should return when `session.Context()` is done.
+ // If not, will raise `ErrRebalanceInProgress` or `read tcp <ip>:<port>: i/o timeout` when kafka rebalance. see:
+ // https://github.com/Shopify/sarama/issues/1192
+ case <-session.Context().Done():
return nil
}
+ }
+}
func (consumer *consumer) doCallback(session sarama.ConsumerGroupSession, message *sarama.ConsumerMessage) error {
consumer.k.logger.Debugf("Processing Kafka message: %s/%d/%d [key=%s]", message.Topic, message.Partition, message.Offset, asBase64String(message.Key))
