Updating offset correctly (#17)

msales · Aug 3, 2018 · 114359b · 114359b
1 parent 3557fea
commit 114359b
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/kafka/source.go b/kafka/source.go
@@ -162,6 +162,8 @@ func (s *Source) markState(msg *sarama.ConsumerMessage) []marker {
 		s.state[msg.Topic] = partitions
 	}
 
+	partitions[msg.Partition] = msg.Offset
+
 	var markers []marker
 	for topic, partitions := range s.state {
 		for partition, offset := range partitions {