From b4ddebdac0f1d97b72fe66c1e1feb2c88346c73d Mon Sep 17 00:00:00 2001
From: Jeroen Rinzema <post@jeroen.no>
Date: Fri, 29 Nov 2019 09:46:04 +0100
Subject: [PATCH] fix: #28 mark as done on retry

---
 dialects/kafka/consumer/group.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/dialects/kafka/consumer/group.go b/dialects/kafka/consumer/group.go
index 3bf4a4c..b4fffe6 100644
--- 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
 			}