After a rebalance, the receiver keeps receiving messages until it errors out

[nbajaj90]

After a rebalance, the receiver keeps receiving messages until its error out, and most of these messages fails to commit. Do we need to fix the ConsumeClaim to return once the consumer session ends (because of rebalance), so that the receiver won't keep receiving messages after rebalance.

To fix here:

sdk-go/protocol/kafka_sarama/v2/receiver.go

Line 55 in 6784a56

func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

References:
IBM/sarama#2118
https://github.com/Shopify/sarama/blob/5e2c2ef0e429f895c86152189f625bfdad7d3452/examples/consumergroup/main.go

[nbajaj90]

Expecting something like:

func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for {
		select {
		case msg, ok := <-claim.Messages():
			if !ok {
				return nil
			}
			m := NewMessageFromConsumerMessage(msg)

			r.incoming <- msgErr{
				msg: binding.WithFinish(m, func(err error) {
					if protocol.IsACK(err) {
						session.MarkMessage(msg, "")
					}
				}),
			}
		case <-session.Context().Done():
			return nil
		}
	}
}