fix: #28 mark as done on retry

jeroenrinzema · Nov 29, 2019 · b4ddebd · b4ddebd
1 parent 0cb9ea8
commit b4ddebd
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/dialects/kafka/consumer/group.go b/dialects/kafka/consumer/group.go
@@ -86,6 +86,7 @@ func (handle *GroupHandle) ConsumeClaim(session sarama.ConsumerGroupSession, cla
 			if err == ErrRetry {
 				// Mark the message to be consumed again
 				session.ResetOffset(message.Topic, message.Partition, message.Offset, "")
+				handle.consumptions.Done()
 				return
 			}