Add CommitInterval to kafka new reader

deflect-ca · Dec 14, 2023 · 9fe78b2 · 9fe78b2
1 parent 2f06def
commit 9fe78b2
Showing 1 changed file with 6 additions and 5 deletions.
diff --git a/internal/kafka.go b/internal/kafka.go
@@ -89,11 +89,12 @@ func RunKafkaReader(
 	// XXX this infinite loop is so we reconnect if we get dropped.
 	for {
 		r := kafka.NewReader(kafka.ReaderConfig{
-			Brokers:     config.KafkaBrokers,
-			GroupID:     uuid.New().String(),
-			StartOffset: kafka.LastOffset,
-			Topic:       config.KafkaCommandTopic,
-			Dialer:      getDialer(config),
+			Brokers:        config.KafkaBrokers,
+			GroupID:        uuid.New().String(),
+			StartOffset:    kafka.LastOffset,
+			Topic:          config.KafkaCommandTopic,
+			Dialer:         getDialer(config),
+			CommitInterval: time.Second * 10,
 		})
 		defer r.Close()