set MaxMessageBytes for sarama producer (#588)

* set MaxMessageBytes for sarama producer * fix lint emptyStringTest * small fix in kafka params doc
ozontech · Feb 29, 2024 · 8f3cbe4 · 8f3cbe4
1 parent 5a3d4d1
commit 8f3cbe4
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 6 deletions.
diff --git a/e2e/kafka_auth/kafka_auth.go b/e2e/kafka_auth/kafka_auth.go
@@ -86,10 +86,11 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 		testFns := []func(){
 			func() {
 				config := &kafka_out.Config{
-					Brokers:      c.Brokers,
-					DefaultTopic: outTopic,
-					ClientID:     "test-auth-out",
-					BatchSize_:   10,
+					Brokers:          c.Brokers,
+					DefaultTopic:     outTopic,
+					ClientID:         "test-auth-out",
+					BatchSize_:       10,
+					MaxMessageBytes_: 1000000,
 				}
 				if tt.sasl.Enabled {
 					config.SaslEnabled = true

diff --git a/fd/util.go b/fd/util.go
@@ -149,7 +149,7 @@ func extractConditions(condJSON *simplejson.Json) (pipeline.MatchConditions, err
 		}
 
 		if value, ok := obj.(string); ok {
-			if len(value) > 0 && value[0] == '/' {
+			if value != "" && value[0] == '/' {
 				r, err := cfg.CompileRegex(value)
 				if err != nil {
 					return nil, fmt.Errorf("can't compile regexp %s: %w", value, err)

diff --git a/plugin/action/convert_utf8_bytes/convert_utf8_bytes.go b/plugin/action/convert_utf8_bytes/convert_utf8_bytes.go
@@ -170,7 +170,7 @@ func (p *Plugin) convert(node *insaneJSON.Node) {
 	p.buf = append(p.buf, nodeStr[:idx]...)
 	nodeStr = nodeStr[idx+1:]
 
-	for len(nodeStr) > 0 {
+	for nodeStr != "" {
 		ch := nodeStr[0]
 		switch ch {
 		case '\\':

diff --git a/plugin/output/kafka/README.md b/plugin/output/kafka/README.md
@@ -57,6 +57,13 @@ After this timeout the batch will be sent even if batch isn't full.
 
 <br>
 
+**`max_message_bytes`** *`cfg.Expression`* *`default=1000000`* 
+
+The maximum permitted size of a message.
+Should be set equal to or smaller than the broker's `message.max.bytes`.
+
+<br>
+
 **`retry`** *`int`* *`default=10`* 
 
 Retries of insertion. If File.d cannot insert for this number of attempts,

diff --git a/plugin/output/kafka/kafka.go b/plugin/output/kafka/kafka.go
@@ -96,6 +96,13 @@ type Config struct {
 	BatchFlushTimeout  cfg.Duration `json:"batch_flush_timeout" default:"200ms" parse:"duration"` // *
 	BatchFlushTimeout_ time.Duration
 
+	// > @3@4@5@6
+	// >
+	// > The maximum permitted size of a message.
+	// > Should be set equal to or smaller than the broker's `message.max.bytes`.
+	MaxMessageBytes  cfg.Expression `json:"max_message_bytes" default:"1000000" parse:"expression"` // *
+	MaxMessageBytes_ int
+
 	// > @3@4@5@6
 	// >
 	// > Retries of insertion. If File.d cannot insert for this number of attempts,
@@ -321,6 +328,7 @@ func NewProducer(c *Config, l *zap.SugaredLogger) sarama.SyncProducer {
 		config.Net.TLS.Config = tlsCfg.Build()
 	}
 
+	config.Producer.MaxMessageBytes = c.MaxMessageBytes_
 	config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
 	config.Producer.Flush.Messages = c.BatchSize_
 	// kafka plugin itself cares for flush frequency, but we are using batcher so disable it.