increment offset on control message

ricardo-ch · May 25, 2018 · 525bef5 · 525bef5
1 parent 3c763ff
commit 525bef5
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/consumer.go b/consumer.go
@@ -580,6 +580,9 @@ func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*Consu
 	messages := []*ConsumerMessage{}
 	for _, records := range block.RecordsSet {
 		if control, err := records.isControl(); err != nil || control {
+			if err == nil {
+				child.offset = records.RecordBatch.FirstOffset + records.RecordBatch.Records[0].OffsetDelta + 1
+			}
 			continue
 		}