feat: rdkafka max_send_limit_bytes

Signed-off-by: Peter Wilcsinszky <[email protected]>

Author: pepov

charts/logging-operator/crds/logging.banzaicloud.io_clusteroutputs.yaml

@@ -3215,6 +3215,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:
@@ -10616,6 +10618,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:

charts/logging-operator/crds/logging.banzaicloud.io_outputs.yaml

@@ -3211,6 +3211,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:
@@ -9886,6 +9888,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:

config/crd/bases/logging.banzaicloud.io_clusteroutputs.yaml

@@ -3215,6 +3215,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:
@@ -10616,6 +10618,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:

config/crd/bases/logging.banzaicloud.io_outputs.yaml

@@ -3211,6 +3211,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:
@@ -9886,6 +9888,8 @@ spec:
                             x-kubernetes-map-type: atomic
                         type: object
                     type: object
+                  max_send_limit_bytes:
+                    type: integer
                   max_send_retries:
                     type: integer
                   message_key_key:

docs/configuration/plugins/outputs/kafka.md

@@ -146,6 +146,12 @@ Default: nil
 ### keytab (*secret.Secret, optional) {#kafka-keytab}
 
 
+### max_send_limit_bytes (int, optional) {#kafka-max_send_limit_bytes}
+
+Max byte size to send message to avoid MessageSizeTooLarge. Messages over the limit will be dropped
+
+Default: no limit
+
 ### max_send_retries (int, optional) {#kafka-max_send_retries}
 
 Number of times to retry sending of messages to a leader

pkg/sdk/logging/model/output/kafka.go

@@ -111,6 +111,8 @@ type KafkaOutputConfig struct {
 	ScramMechanism string `json:"scram_mechanism,omitempty"`
 	// Number of times to retry sending of messages to a leader (default: 1)
 	MaxSendRetries int `json:"max_send_retries,omitempty"`
+	// Max byte size to send message to avoid MessageSizeTooLarge. Messages over the limit will be dropped (default: no limit)
+	MaxSendLimitBytes int `json:"max_send_limit_bytes,omitempty"`
 	// The number of acks required per request (default: -1).
 	RequiredAcks int `json:"required_acks,omitempty"`
 	// How long the producer waits for acks. The unit is seconds (default: nil => Uses default of ruby-kafka library)