chore(kafka): batch processor config (#429)

ethpandaops · Feb 9, 2025 · ceb2a9e · ceb2a9e
1 parent 95101a0
commit ceb2a9e
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 12 deletions.
diff --git a/pkg/output/kafka/config.go b/pkg/output/kafka/config.go
@@ -6,18 +6,21 @@ import (
 )
 
 type Config struct {
-	Brokers         string              `yaml:"brokers"`
-	Topic           string              `yaml:"topic"`
-	TLS             bool                `yaml:"tls" default:"false"`
-	TLSClientConfig TLSClientConfig     `yaml:"tlsClientConfig"`
-	MaxQueueSize    int                 `yaml:"maxQueueSize" default:"51200"`
-	FlushFrequency  time.Duration       `yaml:"flushFrequency" default:"10s"`
-	FlushMessages   int                 `yaml:"flushMessages" default:"500"`
-	FlushBytes      int                 `yaml:"flushBytes" default:"1000000"`
-	MaxRetries      int                 `yaml:"maxRetries" default:"3"`
-	Compression     CompressionStrategy `yaml:"compression" default:"none"`
-	RequiredAcks    RequiredAcks        `yaml:"requiredAcks" default:"leader"`
-	Partitioning    PartitionStrategy   `yaml:"partitioning" default:"none"`
+	Brokers            string              `yaml:"brokers"`
+	Topic              string              `yaml:"topic"`
+	TLS                bool                `yaml:"tls" default:"false"`
+	TLSClientConfig    TLSClientConfig     `yaml:"tlsClientConfig"`
+	MaxQueueSize       int                 `yaml:"maxQueueSize" default:"51200"`
+	BatchTimeout       time.Duration       `yaml:"batchTimeout" default:"5s"`
+	MaxExportBatchSize int                 `yaml:"maxExportBatchSize" default:"512"`
+	Workers            int                 `yaml:"workers" default:"5"`
+	FlushFrequency     time.Duration       `yaml:"flushFrequency" default:"10s"`
+	FlushMessages      int                 `yaml:"flushMessages" default:"500"`
+	FlushBytes         int                 `yaml:"flushBytes" default:"1000000"`
+	MaxRetries         int                 `yaml:"maxRetries" default:"3"`
+	Compression        CompressionStrategy `yaml:"compression" default:"none"`
+	RequiredAcks       RequiredAcks        `yaml:"requiredAcks" default:"leader"`
+	Partitioning       PartitionStrategy   `yaml:"partitioning" default:"none"`
 }
 
 type TLSClientConfig struct {

diff --git a/pkg/output/kafka/kafka.go b/pkg/output/kafka/kafka.go
@@ -42,7 +42,11 @@ func New(name string, config *Config, log logrus.FieldLogger, filterConfig *xatu
 		exporter,
 		xatu.ImplementationLower()+"_output_"+SinkType+"_"+name,
 		log,
+		processor.WithMaxQueueSize(config.MaxQueueSize),
+		processor.WithBatchTimeout(config.BatchTimeout),
+		processor.WithMaxExportBatchSize(config.MaxExportBatchSize),
 		processor.WithShippingMethod(shippingMethod),
+		processor.WithWorkers(config.Workers),
 	)
 	if err != nil {
 		return nil, err