Changed kafka batchtimeout

netobserv · Jun 14, 2022 · 5626cf0 · 5626cf0
1 parent 102ad5b
commit 5626cf0
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/pkg/pipeline/encode/encode_kafka.go b/pkg/pipeline/encode/encode_kafka.go
@@ -106,6 +106,9 @@ func NewEncodeKafka(params config.StageParam) (Encoder, error) {
 		WriteTimeout: time.Duration(writeTimeoutSecs) * time.Second,
 		BatchSize:    jsonEncodeKafka.BatchSize,
 		BatchBytes:   jsonEncodeKafka.BatchBytes,
+		// Temporary fix may be we should implement a batching systems
+		// https://github.com/segmentio/kafka-go/issues/326#issuecomment-519375403
+		BatchTimeout: 10 * time.Millisecond,
 	}
 
 	return &encodeKafka{