fix: Add Kafka buffer size limit to prevent DataNode OOM (#44106)

issue: https://github.com/milvus-io/milvus/issues/44105

- I have added support to set this property
**queued.max.messages.kbytes** in kafka consumers from the user side.
- It limits the size (in KB) of the consumer’s local message queue
(buffer) where messages are temporarily stored after being fetched from
Kafka but before your application actually processes them

---------

Signed-off-by: Nischay Yadav <Nischay.Yadav@ibm.com>

configs/milvus.yaml

@@ -259,6 +259,7 @@ pulsar:
 #     tlsCaCert:  # file or directory path to CA certificate(s) for verifying the broker's key
 #     tlsKeyPassword:  # private key passphrase for use with ssl.key.location and set_ssl_cert(), if any
 #   readTimeout: 10
+#   queuedmaxkbytes: 100000
 
 rocksmq:
   # Prefix of the key to where Milvus stores data in RocksMQ.

pkg/mq/msgstream/mqwrapper/kafka/kafka_client.go

@@ -86,6 +86,10 @@ func GetBasicConfig(config *paramtable.KafkaConfig) kafka.ConfigMap {
 		kafkaConfig.SetKey("security.protocol", config.SecurityProtocol.GetValue())
 	}
 
+	if config.QueuedMessagesKbytes.GetValue() != "" {
+		kafkaConfig.SetKey("queued.max.messages.kbytes", config.QueuedMessagesKbytes.GetValue())
+	}
+
 	if config.SaslUsername.GetValue() != "" && config.SaslPassword.GetValue() != "" {
 		kafkaConfig.SetKey("sasl.mechanisms", config.SaslMechanisms.GetValue())
 		kafkaConfig.SetKey("sasl.username", config.SaslUsername.GetValue())

pkg/util/paramtable/service_param.go

@@ -1093,6 +1093,7 @@ type KafkaConfig struct {
 	ConsumerExtraConfig  ParamGroup `refreshable:"false"`
 	ProducerExtraConfig  ParamGroup `refreshable:"false"`
 	ReadTimeout          ParamItem  `refreshable:"true"`
+	QueuedMessagesKbytes ParamItem  `refreshable:"false"`
 }
 
 func (k *KafkaConfig) Init(base *BaseTable) {
@@ -1197,6 +1198,14 @@ func (k *KafkaConfig) Init(base *BaseTable) {
 		Export:       true,
 	}
 	k.ReadTimeout.Init(base.mgr)
+
+	k.QueuedMessagesKbytes = ParamItem{
+		Key:          "kafka.queuedmaxkbytes",
+		DefaultValue: "100000", // 100MB in kilo bytes
+		Version:      "2.1.0",
+		Export:       true,
+	}
+	k.QueuedMessagesKbytes.Init(base.mgr)
 }
 
 // /////////////////////////////////////////////////////////////////////////////