Patch sarama_kafka rebalance fix

Issue: cloudevents#817 Issue Explanation: IBM/sarama#2118 Fix reference: https://github.com/Shopify/sarama/blob/5e2c2ef0e429f895c86152189f625bfdad7d3452/examples/consumergroup/main.go#L177 Signed-off-by: nbajaj90 <[email protected]>
nbajaj90 · Nov 3, 2022 · 01c4f13 · 01c4f13
1 parent 6784a56
commit 01c4f13
Showing 1 changed file with 17 additions and 11 deletions.
diff --git a/protocol/kafka_sarama/v2/receiver.go b/protocol/kafka_sarama/v2/receiver.go
@@ -53,19 +53,25 @@ func (r *Receiver) Close(context.Context) error {
 }
 
 func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
-	for message := range claim.Messages() {
-		msg := message
-		m := NewMessageFromConsumerMessage(msg)
-
-		r.incoming <- msgErr{
-			msg: binding.WithFinish(m, func(err error) {
-				if protocol.IsACK(err) {
-					session.MarkMessage(msg, "")
-				}
-			}),
+	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
 		}
 	}
-	return nil
 }
 
 func (r *Receiver) Receive(ctx context.Context) (binding.Message, error) {