context.Done() may never reach if waiting on r.incoming <- msgErr

nbajaj90 · nbajaj90 · commit 6aa2742197a8 · 2023-09-13T17:54:50.000+05:30
Signed-off-by: nbajaj90 &lt;nbajaj90@gmail.com&gt;
diff --git a/protocol/kafka_sarama/v2/receiver.go b/protocol/kafka_sarama/v2/receiver.go
@@ -66,15 +66,23 @@ func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 				return nil
 			}
 			m := NewMessageFromConsumerMessage(msg)
-
-			r.incoming <- msgErr{
+			msgErrObj := msgErr{
 				msg: binding.WithFinish(m, func(err error) {
 					if protocol.IsACK(err) {
 						session.MarkMessage(msg, "")
 					}
 				}),
 			}
 
+			// Need to use select clause here, otherwise r.incoming <- msgErrObj can become a blocking operation,
+			// resulting in never reaching outside block's case <-session.Context().Done()
+			select {
+			case r.incoming <- msgErrObj:
+				// do nothing
+			case <-session.Context().Done():
+				return nil
+			}
+
 		// 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
