gomaxprocs*1 default for kafka consumer max_concurrent_fetches

ozontech · Aug 19, 2024 · 64c939d · 64c939d
1 parent 2f880de
commit 64c939d
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/plugin/input/kafka/client.go b/plugin/input/kafka/client.go
@@ -16,7 +16,7 @@ func NewClient(c *Config, l *zap.Logger, s Consumer) *kgo.Client {
 		kgo.ConsumeTopics(c.Topics...),
 		kgo.FetchMaxWait(c.ConsumerMaxWaitTime_),
 		kgo.AutoCommitMarks(),
-		kgo.MaxConcurrentFetches(c.MaxConcurrentFetches),
+		kgo.MaxConcurrentFetches(c.MaxConcurrentFetches_),
 		kgo.FetchMaxBytes(c.FetchMaxBytes_),
 		kgo.FetchMinBytes(c.FetchMinBytes_),
 		kgo.AutoCommitInterval(c.AutoCommitInterval_),

diff --git a/plugin/input/kafka/kafka.go b/plugin/input/kafka/kafka.go
@@ -102,7 +102,8 @@ type Config struct {
 	// > MaxConcurrentFetches sets the maximum number of fetch requests to allow in
 	// > flight or buffered at once, overriding the unbounded (i.e. number of
 	// > brokers) default.
-	MaxConcurrentFetches int `json:"max_concurrent_fetches" default:"0"` // *
+	MaxConcurrentFetches  cfg.Expression `json:"max_concurrent_fetches" default:"gomaxprocs*1" parse:"expression"` // *
+	MaxConcurrentFetches_ int
 
 	// > @3@4@5@6
 	// >