Merge pull request #24 from tryfix/global_consumer_sync_stuck

Global table sync gets stuck fixed
tryfix · May 25, 2022 · 09904d1 · 09904d1
2 parents da45145 + 70a614d
commit 09904d1
Showing 1 changed file with 5 additions and 2 deletions.
diff --git a/consumer/partition_consumer.go b/consumer/partition_consumer.go
@@ -214,8 +214,11 @@ MainLoop:
 				Headers:   msg.Headers,
 			}
 
-			//if highWatermark == 0 || highWatermark-1 == msg.Offset {
-			if msg.Offset == highWatermark-1 {
+			if msg.Offset == highWatermark-1 ||
+				// TODO this is a workaround to avoid the last message
+				// being a control record(https://kafka.apache.org/documentation/#controlbatch)
+				// when using transactional batches
+				msg.Offset == highWatermark-2 {
 				c.consumerEvents <- &PartitionEnd{
 					tps: []TopicPartition{{
 						Topic:     msg.Topic,