rm kafka input option max_concurrent_fetches

plugin/input/kafka/README.md

@@ -59,14 +59,6 @@ The number of unprocessed messages in the buffer that are loaded in the backgrou
 
 <br>
 
-**`max_concurrent_fetches`** *`int`* *`default=0`* 
-
-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.
-
-<br>
-
 **`fetch_max_bytes`** *`cfg.Expression`* *`default=52428800`* 
 
 FetchMaxBytes (fetch.max.bytes) sets the maximum amount of bytes a broker will try to send during a fetch

plugin/input/kafka/client.go

@@ -2,6 +2,7 @@ package kafka
 
 import (
 	"context"
+	"runtime"
 	"time"
 
 	"github.com/ozontech/file.d/cfg"
@@ -16,7 +17,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(runtime.GOMAXPROCS(0)),
 		kgo.FetchMaxBytes(c.FetchMaxBytes_),
 		kgo.FetchMinBytes(c.FetchMinBytes_),
 		kgo.AutoCommitInterval(c.AutoCommitInterval_),

plugin/input/kafka/kafka.go

@@ -97,13 +97,6 @@ type Config struct {
 	// > The number of unprocessed messages in the buffer that are loaded in the background from kafka. (max.poll.records)
 	ChannelBufferSize int `json:"channel_buffer_size" default:"256"` // *
 
-	// > @3@4@5@6
-	// >
-	// > 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"` // *
-
 	// > @3@4@5@6
 	// >
 	// > FetchMaxBytes (fetch.max.bytes) sets the maximum amount of bytes a broker will try to send during a fetch