Question Regarding ConsumerGroup's Consume() method

[zzhangsg]

Problem Description

Hi there, my team is looking for some help with the usage of ConsumerGroup's Consume() method, currently we have a for loop

        for {
		select {
		case <-shutdownChan:
			log.Info("shutdown channel closed")
			return nil
		case <-ctx.Done():
			log.Error("context in Consume has been cancelled")
			return ctx.Err()
		default:
			if err := c.group.Consume(ctx, []string{c.KafkaConfig.Topic}, handler); err != nil {
				log.WithError(err).Error("group.Consume error")
				return err
			}
		}
	}

We are following what is recommended in the method comments and putting it inside an infinite loop but we are wondering what is the best practice to handle the errors returned from Consume() and would like to hear some insights from the maintainers, should we have the line return err to return any errors OR should we not return it and just log the errors, let the for loop to do the retry magic for us?

IMO not returning errors and having the for loop to retry should be OK to handle most if not all of the errors we will be facing? Would like to hear more inputs, thanks in advance!

[zzhangsg]

@eapache @bai @d1egoaz @dim @dnwe
Hi there, could any gentleman please help take a look? Appreciate it!

[d1egoaz]

Usually the errors returned by .Consume are not recoverable, newSession will try the best effort, up to c.config.Consumer.Group.Rebalance.Retry.Max attempts to get a valid consumer group, assuming the preconditions are correct (brokers reachables, timeout configured correctly, topics exists, ssl certificates are valid and aren't expired, etc)

https://github.com/Shopify/sarama/blob/65f0fec86aabe011db77ad641d31fddf14f3ca41/consumer_group.go#L205

In my experience, I've seen that most cases devs just return the error to close the consumer loop, and try to recreate NewConsumerGroup, which basically tries to recreate the client connection, you could potentially have a circuit breaker, or some max attempts before giving up (panic, send alerts about service down, etc), but most of the times I've seen that the error is just propagated (in some cases it'll trigger the process to die, and be recreated (via kubernetes)), but all cases are different, this depends a lot on you infrastructure/arquitecture.