Merge pull request #30 from prOOrc/fix/rebalancing

fix/rebalancing
ThreeDotsLabs · Jun 27, 2024 · 2664bae · 2664bae
2 parents ba31b43 + 4138e3a
commit 2664bae
Showing 1 changed file with 11 additions and 1 deletion.
diff --git a/pkg/kafka/subscriber.go b/pkg/kafka/subscriber.go
@@ -584,7 +584,17 @@ ResendLoop:
 		select {
 		case <-msg.Acked():
 			if sess != nil {
-				sess.MarkMessage(kafkaMsg, "")
+				if sess.Context().Err() == nil {
+					sess.MarkMessage(kafkaMsg, "")
+				} else {
+					logFields := receivedMsgLogFields.Add(
+						watermill.LogFields{
+							"err": sess.Context().Err().Error(),
+						},
+					)
+					h.logger.Trace("Closing, session ctx cancelled before ack", logFields)
+					return nil
+				}
 			}
 			h.logger.Trace("Message Acked", receivedMsgLogFields)
 			break ResendLoop