Commit synchronously

practo · Aug 9, 2020 · 7213c65 · 7213c65
1 parent 345ba76
commit 7213c65
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/redshiftsink/pkg/consumer/sarama.go b/redshiftsink/pkg/consumer/sarama.go
@@ -46,7 +46,9 @@ func (c saramaConsumer) ConsumeClaim(
 		klog.Infof("Message claimed: value = %s, timestamp = %v, topic = %s",
 			string(message.Value), message.Timestamp, message.Topic)
 		session.MarkMessage(message, "")
+		session.Commit()
 	}
+
 	klog.V(4).Info("All messages were consumed, exiting consumerClaim.")
 
 	return nil