fix: Kafka commit session end of consume claim

Signed-off-by: Alex Collins <[email protected]>
argoproj-labs · Sep 20, 2021 · 40eac37 · 40eac37
1 parent a618b38
commit 40eac37
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/runner/sidecar/source/kafka/handler.go b/runner/sidecar/source/kafka/handler.go
@@ -39,6 +39,7 @@ func (h *handler) Cleanup(sess sarama.ConsumerGroupSession) error {
 func (h *handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
 	ctx := sess.Context()
 	logger.Info("starting consuming claim", "partition", claim.Partition())
+	defer sess.Commit()
 	for msg := range claim.Messages() {
 		if err := h.processMessage(ctx, msg); err != nil {
 			logger.Error(err, "failed to process message")

diff --git a/test/kafka-stress/test-results.json b/test/kafka-stress/test-results.json
@@ -1,12 +1,12 @@
 {
-  "TestKafkaSinkStress/.tps": 750,
+  "TestKafkaSinkStress/.tps": 600,
   "TestKafkaSinkStress/N=10,messageSize=100.tps": 200,
   "TestKafkaSinkStress/N=10,messageSize=1000.tps": 150,
   "TestKafkaSinkStress/N=50000.tps": 750,
   "TestKafkaSinkStress/async=true.tps": 400,
   "TestKafkaSinkStress/messageSize=1000.tps": 300,
   "TestKafkaSinkStress/replicas=2.tps": 400,
-  "TestKafkaSourceStress/.tps": 3450,
+  "TestKafkaSourceStress/.tps": 1300,
   "TestKafkaSourceStress/N=10,messageSize=100.tps": 450,
   "TestKafkaSourceStress/N=10,messageSize=1000.tps": 650,
   "TestKafkaSourceStress/N=50000.tps": 3150,