Revert "rm kafka input option max_concurrent_fetches"

This reverts commit 216d4cb.

plugin/input/kafka/README.md

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

plugin/input/kafka/kafka.go

@@ -97,6 +97,13 @@ 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