[exporter/kafka] Add producer.max_broker_write_bytes config option (#49007)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

The maximum size of a single write to a broker was previously fixed at
the underlying franz-go default of 100 MiB and could not be configured.
As a result, setting `producer.max_message_bytes` above 100 MiB passed
configuration validation but caused the collector to fail on startup
with an unrecoverable error.

The new `producer.max_broker_write_bytes` setting (default 104857600,
i.e. 100 MiB) exposes this limit. To send messages larger than 100 MiB,
raise it so it is greater than or equal to `max_message_bytes`.
Configuration is now validated up front: the collector reports a clear
error if `max_broker_write_bytes` is below the 100 MiB minimum or
smaller than `max_message_bytes`, rather than failing at runtime.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes #47492

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Added unit tests.

<!--Describe the documentation added.-->
#### Documentation

Updated the README.md with this new option.

<!--Authorship attestation. See AGENTS.md for details. AI agents must
not check this box on behalf
of the user; the human author must check it themselves before the PR is
ready for review.-->
#### Authorship

- [X] I, a human, wrote this pull request description myself.

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Signed-off-by: Paulo Dias <paulodias.gm@gmail.com>

Author: paulojmdias

.chloggen/47492-kafkaexporter-max-broker-write-bytes.yaml

@@ -0,0 +1,38 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: exporter/kafka
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `producer.max_broker_write_bytes` config
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47492]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  The maximum size of a single write to a broker was previously fixed at the underlying
+  franz-go default of 100 MiB and could not be configured. As a result, setting
+  `producer.max_message_bytes` above 100 MiB passed configuration validation but caused the
+  collector to fail on startup with an unrecoverable error ("max broker write bytes ... is
+  erroneously less than max record batch bytes ...").
+
+  The new `producer.max_broker_write_bytes` setting (default 104857600, i.e. 100 MiB) exposes
+  this limit. To send messages larger than 100 MiB, raise it so it is greater than or equal to
+  `max_message_bytes`. Configuration is now validated up front: the collector reports a clear
+  error if `max_broker_write_bytes` is below the 100 MiB minimum or smaller than
+  `max_message_bytes`, rather than failing at runtime.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/kafkaexporter/README.md

@@ -105,6 +105,7 @@ The following settings can be optionally configured:
     - `requests_per_second` is the average number of requests per seconds.
 - `producer`
   - `max_message_bytes` (default = 1000000) the maximum permitted size of a message in bytes, calculated before compression.
+  - `max_broker_write_bytes` (default = 104857600) the maximum bytes the producer will write to a broker in a single request. Must be greater than or equal to `max_message_bytes`. Increase this when raising `max_message_bytes` above 100 MiB.
   - `required_acks` (default = 1) controls when a message is regarded as transmitted. <https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks>
   - `compression` (default = 'none') the compression used when producing messages to kafka. The options are: `none`, `gzip`, `snappy`, `lz4`, and `zstd` <https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#compression-type>
   - `compression_params`

internal/kafka/franz_client.go

@@ -70,6 +70,7 @@ func NewFranzSyncProducer(
 		kgo.ProducerBatchCompression(codec),
 		kgo.ProducerLinger(cfg.Linger),
 		kgo.ProducerBatchMaxBytes(int32(cfg.MaxMessageBytes)),
+		kgo.BrokerMaxWriteBytes(int32(cfg.MaxBrokerWriteBytes)),
 		kgo.MaxBufferedRecords(cfg.FlushMaxMessages),
 	)...)
 	if err != nil {

pkg/kafka/configkafka/config.go

@@ -6,6 +6,7 @@ package configkafka // import "github.com/open-telemetry/opentelemetry-collector
 import (
 	"errors"
 	"fmt"
+	"math"
 	"strings"
 	"time"
 
@@ -268,10 +269,26 @@ type AutoCommitConfig struct {
 	Interval time.Duration `mapstructure:"interval"`
 }
 
+// franzGoMinBrokerWriteBytes is franz-go's hardcoded 100 MiB floor for
+// kgo.BrokerMaxWriteBytes: values below it are rejected by franz-go at client
+// construction. It is also franz-go's default for that option, so it doubles as
+// the default for ProducerConfig.MaxBrokerWriteBytes, preserving the prior
+// (non-configurable) behavior when left unset.
+const franzGoMinBrokerWriteBytes = 100 << 20 // 104857600
+
 type ProducerConfig struct {
-	// Maximum message bytes the producer will accept to produce (default 1000000)
+	// MaxMessageBytes is the maximum message bytes the producer will accept to
+	// produce. It must be less than or equal to MaxBrokerWriteBytes, and must
+	// fit in an int32 as it maps to franz-go's kgo.ProducerBatchMaxBytes.
+	// (default 1000000)
 	MaxMessageBytes int `mapstructure:"max_message_bytes"`
 
+	// MaxBrokerWriteBytes is the maximum bytes the producer will write to a
+	// broker in a single request. It must be >= MaxMessageBytes. Maps to
+	// franz-go's kgo.BrokerMaxWriteBytes, whose default (and minimum accepted
+	// value) is 100 MiB. (default 104857600)
+	MaxBrokerWriteBytes int `mapstructure:"max_broker_write_bytes"`
+
 	// RequiredAcks holds the number acknowledgements required before producing
 	// returns successfully. See:
 	// https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks
@@ -309,6 +326,7 @@ type ProducerConfig struct {
 func NewDefaultProducerConfig() ProducerConfig {
 	return ProducerConfig{
 		MaxMessageBytes:        1000000,
+		MaxBrokerWriteBytes:    franzGoMinBrokerWriteBytes,
 		RequiredAcks:           WaitForLocal,
 		Compression:            "none",
 		FlushMaxMessages:       10000,
@@ -335,6 +353,32 @@ func (c ProducerConfig) Validate() error {
 	if c.MaxMessageBytes < 0 {
 		return fmt.Errorf("max_message_bytes (%d) must be non-negative", c.MaxMessageBytes)
 	}
+	// Both limits are passed to franz-go as int32, so reject anything that would
+	// overflow on conversion and silently become a negative/invalid size.
+	if c.MaxMessageBytes > math.MaxInt32 {
+		return fmt.Errorf("max_message_bytes (%d) must not exceed %d", c.MaxMessageBytes, math.MaxInt32)
+	}
+	if c.MaxBrokerWriteBytes < 0 {
+		return fmt.Errorf("max_broker_write_bytes (%d) must be non-negative", c.MaxBrokerWriteBytes)
+	}
+	if c.MaxBrokerWriteBytes < franzGoMinBrokerWriteBytes {
+		return fmt.Errorf(
+			"max_broker_write_bytes (%d) must be at least %d (%d MiB, franz-go minimum)",
+			c.MaxBrokerWriteBytes,
+			franzGoMinBrokerWriteBytes,
+			franzGoMinBrokerWriteBytes>>20,
+		)
+	}
+	if c.MaxBrokerWriteBytes > math.MaxInt32 {
+		return fmt.Errorf("max_broker_write_bytes (%d) must not exceed %d", c.MaxBrokerWriteBytes, math.MaxInt32)
+	}
+	if c.MaxMessageBytes > c.MaxBrokerWriteBytes {
+		return fmt.Errorf(
+			"max_message_bytes (%d) cannot be greater than max_broker_write_bytes (%d)",
+			c.MaxMessageBytes,
+			c.MaxBrokerWriteBytes,
+		)
+	}
 	if c.FlushMaxMessages < 1 {
 		return fmt.Errorf("flush_max_messages (%d) must be at least 1", c.FlushMaxMessages)
 	}

pkg/kafka/configkafka/config.schema.yaml

@@ -194,8 +194,11 @@ $defs:
         description: Linger controls the linger time for the producer. (default 10ms).
         type: string
         format: duration
+      max_broker_write_bytes:
+        description: MaxBrokerWriteBytes is the maximum bytes the producer will write to a broker in a single request. It must be >= MaxMessageBytes. Maps to franz-go's kgo.BrokerMaxWriteBytes, whose default (and minimum accepted value) is 100 MiB. (default 104857600)
+        type: integer
       max_message_bytes:
-        description: Maximum message bytes the producer will accept to produce (default 1000000)
+        description: MaxMessageBytes is the maximum message bytes the producer will accept to produce. It must be less than or equal to MaxBrokerWriteBytes, and must fit in an int32 as it maps to franz-go's kgo.ProducerBatchMaxBytes. (default 1000000)
         type: integer
       required_acks:
         description: 'RequiredAcks holds the number acknowledgements required before producing returns successfully. See: https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks Acceptable values are: 0 (NoResponse)   Does not wait for any acknowledgements. 1 (WaitForLocal) Waits for only the leader to write the record to its local log, but does not wait for followers to acknowledge. (default) -1 (WaitForAll)   Waits for all in-sync replicas to acknowledge. In YAML configuration, "all" is accepted as an alias for -1.'

pkg/kafka/configkafka/config_test.go

@@ -4,7 +4,10 @@
 package configkafka // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/kafka/configkafka"
 
 import (
+	"fmt"
+	"math"
 	"path/filepath"
+	"strconv"
 	"testing"
 	"time"
 
@@ -296,6 +299,14 @@ func TestProducerConfig(t *testing.T) {
 				return cfg
 			}(),
 		},
+		"large_message": {
+			expected: func() ProducerConfig {
+				cfg := NewDefaultProducerConfig()
+				cfg.MaxMessageBytes = 209715200
+				cfg.MaxBrokerWriteBytes = 268435456
+				return cfg
+			}(),
+		},
 
 		// Invalid configurations
 		"invalid_compression": {
@@ -313,6 +324,41 @@ func TestProducerConfig(t *testing.T) {
 		"max_message_bytes_negative": {
 			expectedErr: "max_message_bytes (-1000) must be non-negative",
 		},
+		"max_broker_write_bytes_negative": {
+			expectedErr: "max_broker_write_bytes (-1000) must be non-negative",
+		},
+		"max_broker_write_bytes_too_small": {
+			expectedErr: fmt.Sprintf("max_broker_write_bytes (1000) must be at least %d (%d MiB, franz-go minimum)", franzGoMinBrokerWriteBytes, franzGoMinBrokerWriteBytes>>20),
+		},
+		"max_message_bytes_exceeds_broker": {
+			expectedErr: fmt.Sprintf("max_message_bytes (209715200) cannot be greater than max_broker_write_bytes (%d)", franzGoMinBrokerWriteBytes),
+		},
+	})
+}
+
+// TestProducerConfigInt32Overflow verifies that size limits exceeding the int32
+// range used by franz-go are rejected. Values above math.MaxInt32 are only
+// representable on platforms where int is 64-bit, so this is skipped elsewhere.
+func TestProducerConfigInt32Overflow(t *testing.T) {
+	if strconv.IntSize < 64 {
+		t.Skip("int is not wide enough to exceed math.MaxInt32 on this architecture")
+	}
+	// Use a runtime int64 value so the conversion to int is not a compile-time
+	// constant (which would overflow int and fail to build on 32-bit platforms).
+	maxInt32 := int64(math.MaxInt32)
+	overflow := int(maxInt32 + 1)
+
+	t.Run("max_message_bytes", func(t *testing.T) {
+		cfg := NewDefaultProducerConfig()
+		cfg.MaxMessageBytes = overflow
+		require.EqualError(t, cfg.Validate(),
+			fmt.Sprintf("max_message_bytes (%d) must not exceed %d", overflow, math.MaxInt32))
+	})
+	t.Run("max_broker_write_bytes", func(t *testing.T) {
+		cfg := NewDefaultProducerConfig()
+		cfg.MaxBrokerWriteBytes = overflow
+		require.EqualError(t, cfg.Validate(),
+			fmt.Sprintf("max_broker_write_bytes (%d) must not exceed %d", overflow, math.MaxInt32))
 	})
 }
 

pkg/kafka/configkafka/testdata/producer_config.yaml

@@ -25,6 +25,9 @@ kafka/required_acks_all:
   required_acks: all
 kafka/disable_auto_topic_creation:
   allow_auto_topic_creation: false
+kafka/large_message:
+  max_message_bytes: 209715200
+  max_broker_write_bytes: 268435456
 
 # Invalid configurations
 kafka/invalid_compression:
@@ -37,6 +40,12 @@ kafka/flush_max_messages_negative:
   flush_max_messages: -1
 kafka/max_message_bytes_negative:
   max_message_bytes: -1000
+kafka/max_broker_write_bytes_too_small:
+  max_broker_write_bytes: 1000
+kafka/max_message_bytes_exceeds_broker:
+  max_message_bytes: 209715200
+kafka/max_broker_write_bytes_negative:
+  max_broker_write_bytes: -1000
 
 kafka/producer_linger:
   linger: 100ms