From 44ba568044e2a78eebc221f21ddbc299cf2af9c3 Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Wed, 12 Aug 2026 11:44:14 +0200
Subject: [PATCH 1/8] misc: fix broken otel query from upgrade

---
 .../alloy-mixin/dashboards/otel-engine-overview.libsonnet     | 4 ++++
 .../rendered/dashboards/alloy-otel-engine-overview.json       | 4 ++--
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/operations/alloy-mixin/dashboards/otel-engine-overview.libsonnet b/operations/alloy-mixin/dashboards/otel-engine-overview.libsonnet
index 821b827e2c2..74f7cc49a29 100644
--- a/operations/alloy-mixin/dashboards/otel-engine-overview.libsonnet
+++ b/operations/alloy-mixin/dashboards/otel-engine-overview.libsonnet
@@ -431,12 +431,16 @@ local filename = 'alloy-otel-engine-overview.json';
               panel.withQueries([
                 panel.newQuery(
                   expr=|||
+                    sum by(${groupby}) (rate(otelcol_processor_memory_limiter_refused_spans_total{%(groupSelector)s}[$__rate_interval]))
+                    or
                     sum by(${groupby}) (rate(otelcol_processor_refused_spans_total{%(groupSelector)s}[$__rate_interval]))
                   ||| % $._config,
                   legendFormat='{{${groupby}}} spans',
                 ),
                 panel.newQuery(
                   expr=|||
+                    sum by(${groupby}) (rate(otelcol_processor_memory_limiter_refused_metric_points_total{%(groupSelector)s}[$__rate_interval]))
+                    or
                     sum by(${groupby}) (rate(otelcol_processor_refused_metric_points_total{%(groupSelector)s}[$__rate_interval]))
                   ||| % $._config,
                   legendFormat='{{${groupby}}} metric points',
diff --git a/operations/alloy-mixin/rendered/dashboards/alloy-otel-engine-overview.json b/operations/alloy-mixin/rendered/dashboards/alloy-otel-engine-overview.json
index 13c159bcf78..29b67d55abb 100644
--- a/operations/alloy-mixin/rendered/dashboards/alloy-otel-engine-overview.json
+++ b/operations/alloy-mixin/rendered/dashboards/alloy-otel-engine-overview.json
@@ -964,14 +964,14 @@
                "targets": [
                   {
                      "datasource": "${datasource}",
-                     "expr": "sum by(${groupby}) (rate(otelcol_processor_refused_spans_total{cluster=~\"$cluster\", namespace=~\"$namespace\", job=~\"$job\"}[$__rate_interval]))\n",
+                     "expr": "sum by(${groupby}) (rate(otelcol_processor_memory_limiter_refused_spans_total{cluster=~\"$cluster\", namespace=~\"$namespace\", job=~\"$job\"}[$__rate_interval]))\nor\nsum by(${groupby}) (rate(otelcol_processor_refused_spans_total{cluster=~\"$cluster\", namespace=~\"$namespace\", job=~\"$job\"}[$__rate_interval]))\n",
                      "instant": false,
                      "legendFormat": "{{${groupby}}} spans",
                      "range": true
                   },
                   {
                      "datasource": "${datasource}",
-                     "expr": "sum by(${groupby}) (rate(otelcol_processor_refused_metric_points_total{cluster=~\"$cluster\", namespace=~\"$namespace\", job=~\"$job\"}[$__rate_interval]))\n",
+                     "expr": "sum by(${groupby}) (rate(otelcol_processor_memory_limiter_refused_metric_points_total{cluster=~\"$cluster\", namespace=~\"$namespace\", job=~\"$job\"}[$__rate_interval]))\nor\nsum by(${groupby}) (rate(otelcol_processor_refused_metric_points_total{cluster=~\"$cluster\", namespace=~\"$namespace\", job=~\"$job\"}[$__rate_interval]))\n",
                      "instant": false,
                      "legendFormat": "{{${groupby}}} metric points",
                      "range": true

From 38f1698c0af895fe5930a281f2a53aec6b096f0e Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Wed, 12 Aug 2026 11:56:52 +0200
Subject: [PATCH 2/8] expose max_message_bytes and linger in kafka exporter

---
 .../otelcol/otelcol.exporter.kafka.md         | 19 +++--
 .../component/otelcol/exporter/kafka/kafka.go | 21 +++++-
 .../otelcol/exporter/kafka/kafka_test.go      | 69 ++++++++++++++++---
 3 files changed, 92 insertions(+), 17 deletions(-)

diff --git a/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md b/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
index 06e16a74010..74bb7d262ac 100644
--- a/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
+++ b/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
@@ -215,15 +215,22 @@ The `producer` block configures how to retry retrieving metadata when retrieval
 
 The following arguments are supported:
 
-| Name                 | Type     | Description                                         | Default   | Required |
-| -------------------- | -------- | --------------------------------------------------- | --------- | -------- |
-| `compression`        | `string` | The level of compression to use on messages.        | `"none"`  | no       |
-| `flush_max_messages` | `number` | The maximum number of messages in one request.      | `10000`   | no       |
-| `max_message_bytes`  | `number` | The maximum permitted size of a message in bytes.   | `1000000` | no       |
-| `required_acks`      | `number` | Controls when a message is regarded as transmitted. | `1`       | no       |
+| Name                     | Type       | Description                                                                | Default     | Required |
+| ------------------------ | ---------- | -------------------------------------------------------------------------- | ----------- | -------- |
+| `compression`            | `string`   | The level of compression to use on messages.                               | `"none"`    | no       |
+| `flush_max_messages`     | `number`   | The maximum number of messages in one request.                             | `10000`     | no       |
+| `linger`                 | `duration` | How long a topic partition waits for more records before building a request. | `"10ms"`    | no       |
+| `max_broker_write_bytes` | `number`   | The maximum permitted size of a single write to a broker in bytes.         | `104857600` | no       |
+| `max_message_bytes`      | `number`   | The maximum permitted size of a message in bytes.                          | `1000000`   | no       |
+| `required_acks`          | `number`   | Controls when a message is regarded as transmitted.                        | `1`         | no       |
 
 Refer to the [Go sarama documentation][RequiredAcks] for more information on `required_acks`.
 
+`max_broker_write_bytes` must be at least `104857600` (100 MiB), and `max_message_bytes` must be less than or equal to it.
+Raise `max_broker_write_bytes` if you need a `max_message_bytes` larger than the default.
+
+Set `linger` to `"0s"` to send records as soon as they arrive, at the cost of less effective batching.
+
 `compression` could be set to either `none`, `gzip`, `snappy`, `lz4`, or `zstd`.
 Refer to the [Go sarama documentation][CompressionCodec] for more information.
 
diff --git a/internal/component/otelcol/exporter/kafka/kafka.go b/internal/component/otelcol/exporter/kafka/kafka.go
index b1dbe54e524..47d2a9d0337 100644
--- a/internal/component/otelcol/exporter/kafka/kafka.go
+++ b/internal/component/otelcol/exporter/kafka/kafka.go
@@ -175,6 +175,11 @@ type Producer struct {
 	// Maximum message bytes the producer will accept to produce.
 	MaxMessageBytes int `alloy:"max_message_bytes,attr,optional"`
 
+	// MaxBrokerWriteBytes is the maximum bytes the producer will write to a broker
+	// in a single request. Must be greater than or equal to max_message_bytes, and
+	// at least 100 MiB
+	MaxBrokerWriteBytes int `alloy:"max_broker_write_bytes,attr,optional"`
+
 	// RequiredAcks Number of acknowledgements required to assume that a message has been sent.
 	// https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks
 	// The options are:
@@ -198,17 +203,23 @@ type Producer struct {
 
 	// Whether or not to allow automatic topic creation.
 	AllowAutoTopicCreation bool `alloy:"allow_auto_topic_creation,attr,optional"`
+
+	// Linger is how long individual topic partitions wait for more records before
+	// a request is built. Set to "0s" to send records as soon as they arrive.
+	Linger time.Duration `alloy:"linger,attr,optional"`
 }
 
 // Convert converts args into the upstream type.
 func (args Producer) Convert() configkafka.ProducerConfig {
 	cfg := configkafka.NewDefaultProducerConfig()
 	cfg.MaxMessageBytes = args.MaxMessageBytes
+	cfg.MaxBrokerWriteBytes = args.MaxBrokerWriteBytes
 	cfg.RequiredAcks = configkafka.RequiredAcks(args.RequiredAcks)
 	cfg.Compression = args.Compression
 	cfg.CompressionParams = args.CompressionParams.Convert()
 	cfg.FlushMaxMessages = args.FlushMaxMessages
 	cfg.AllowAutoTopicCreation = args.AllowAutoTopicCreation
+	cfg.Linger = args.Linger
 	return cfg
 }
 
@@ -230,6 +241,8 @@ var (
 
 // SetToDefault implements syntax.Defaulter.
 func (args *Arguments) SetToDefault() {
+	producerDefaults := configkafka.NewDefaultProducerConfig()
+
 	*args = Arguments{
 		Brokers:         []string{"localhost:9092"},
 		ClientID:        "otel-collector",
@@ -244,14 +257,16 @@ func (args *Arguments) SetToDefault() {
 			},
 		},
 		Producer: Producer{
-			MaxMessageBytes: 1000000,
-			RequiredAcks:    1,
-			Compression:     "none",
+			MaxMessageBytes:     1000000,
+			MaxBrokerWriteBytes: producerDefaults.MaxBrokerWriteBytes,
+			RequiredAcks:        1,
+			Compression:         "none",
 			CompressionParams: CompressionParams{
 				Level: 0, // Default compression level
 			},
 			FlushMaxMessages:       10000,
 			AllowAutoTopicCreation: true,
+			Linger:                 producerDefaults.Linger,
 		},
 		RecordPartitioner: &RecordPartitionerConfig{
 			StickyKey: &StickyKeyPartitionerConfig{Hasher: "sarama_compat"},
diff --git a/internal/component/otelcol/exporter/kafka/kafka_test.go b/internal/component/otelcol/exporter/kafka/kafka_test.go
index 6fa0b2b1119..d1b2141abf3 100644
--- a/internal/component/otelcol/exporter/kafka/kafka_test.go
+++ b/internal/component/otelcol/exporter/kafka/kafka_test.go
@@ -79,8 +79,7 @@ func TestArguments_UnmarshalAlloy(t *testing.T) {
 				},
 			},
 			Producer: configkafka.ProducerConfig{
-				MaxMessageBytes: 1000000,
-				// Not exposed by Alloy; inherited from the upstream factory default.
+				MaxMessageBytes:     1000000,
 				MaxBrokerWriteBytes: 104857600,
 				RequiredAcks:        1,
 				Compression:         "none",
@@ -89,8 +88,7 @@ func TestArguments_UnmarshalAlloy(t *testing.T) {
 				},
 				FlushMaxMessages:       10000,
 				AllowAutoTopicCreation: true,
-				// Not exposed by Alloy; inherited from the upstream factory default.
-				Linger: 10 * time.Millisecond,
+				Linger:                 10 * time.Millisecond,
 			},
 		}
 	}
@@ -348,8 +346,7 @@ func TestArguments_UnmarshalAlloy(t *testing.T) {
 					},
 				},
 				Producer: configkafka.ProducerConfig{
-					MaxMessageBytes: 2000001,
-					// Not exposed by Alloy; inherited from the upstream factory default.
+					MaxMessageBytes:     2000001,
 					MaxBrokerWriteBytes: 104857600,
 					RequiredAcks:        0,
 					Compression:         "gzip",
@@ -358,8 +355,7 @@ func TestArguments_UnmarshalAlloy(t *testing.T) {
 					},
 					FlushMaxMessages:       101,
 					AllowAutoTopicCreation: true,
-					// Not exposed by Alloy; inherited from the upstream factory default.
-					Linger: 10 * time.Millisecond,
+					Linger:                 10 * time.Millisecond,
 				},
 				IncludeMetadataKeys:                  []string(nil),
 				TopicFromAttribute:                   "my-attr",
@@ -546,3 +542,60 @@ func TestGetSignalType(t *testing.T) {
 		}
 	}
 }
+
+func TestProducerNewFields(t *testing.T) {
+	convert := func(t *testing.T, cfg string) *kafkaexporter.Config {
+		var args kafka.Arguments
+		require.NoError(t, syntax.Unmarshal([]byte(cfg), &args))
+		converted, err := args.Convert()
+		require.NoError(t, err)
+		return converted.(*kafkaexporter.Config)
+	}
+
+	base := `
+		protocol_version = "2.0.0"
+	`
+
+	t.Run("defaults match the upstream factory", func(t *testing.T) {
+		upstream := configkafka.NewDefaultProducerConfig()
+		otelObj := convert(t, base)
+
+		require.Equal(t, upstream.MaxBrokerWriteBytes, otelObj.Producer.MaxBrokerWriteBytes)
+		require.Equal(t, upstream.Linger, otelObj.Producer.Linger)
+	})
+
+	t.Run("configured values are passed through", func(t *testing.T) {
+		otelObj := convert(t, base+`
+			producer {
+				max_broker_write_bytes = 209715200
+				linger                 = "0s"
+			}
+		`)
+
+		require.Equal(t, 209715200, otelObj.Producer.MaxBrokerWriteBytes)
+		require.Equal(t, time.Duration(0), otelObj.Producer.Linger)
+	})
+
+	t.Run("max_message_bytes above the default max_broker_write_bytes is rejected", func(t *testing.T) {
+		var args kafka.Arguments
+		err := syntax.Unmarshal([]byte(base+`
+			producer {
+				max_message_bytes = 209715200
+			}
+		`), &args)
+
+		require.ErrorContains(t, err, "max_message_bytes (209715200) cannot be greater than max_broker_write_bytes (104857600)")
+	})
+
+	t.Run("raising max_broker_write_bytes allows a larger max_message_bytes", func(t *testing.T) {
+		otelObj := convert(t, base+`
+			producer {
+				max_message_bytes      = 209715200
+				max_broker_write_bytes = 209715200
+			}
+		`)
+
+		require.Equal(t, 209715200, otelObj.Producer.MaxMessageBytes)
+		require.Equal(t, 209715200, otelObj.Producer.MaxBrokerWriteBytes)
+	})
+}

From d4afe85fca743febf6a7a932d349aaede0c15dfb Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Wed, 12 Aug 2026 11:57:33 +0200
Subject: [PATCH 3/8] expose group_relabance_strategies in kafka receiver

---
 .../otelcol/otelcol.receiver.kafka.md         | 11 +++-
 .../component/otelcol/receiver/kafka/kafka.go | 58 +++++++++++++----
 .../otelcol/receiver/kafka/kafka_test.go      | 65 +++++++++++++++++++
 3 files changed, 120 insertions(+), 14 deletions(-)

diff --git a/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md b/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md
index 7f72b3c0483..748e77c93c4 100644
--- a/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md
+++ b/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md
@@ -52,6 +52,7 @@ You can use the following arguments with `otelcol.receiver.kafka`:
 | `group_id`                                 | `string`       | Consumer group to consume messages from.                                                                              | `"otel-collector"` | no       |
 | `group_instance_id`                        | `string`       | A unique identifier for the consumer instance within a consumer group.                                                | `""`               | no       |
 | `group_rebalance_strategy`                 | `string`       | The strategy used to assign partitions to consumers within a consumer group.                                          | `"range"`          | no       |
+| `group_rebalance_strategies`               | `list(string)` | The ordered list of strategies to advertise to the group coordinator.                                                 | `[]`               | no       |
 | `heartbeat_interval`                       | `duration`     | The expected time between heartbeats to the consumer coordinator when using Kafka group management.                   | `"3s"`             | no       |
 | `initial_offset`                           | `string`       | Initial offset to use if no offset was previously committed.                                                          | `"latest"`         | no       |
 | `max_fetch_size`                           | `int`          | The maximum number of message bytes to fetch in a request.                                                            | `1048576`          | no       |
@@ -94,9 +95,17 @@ Supported strategies are:
 - `cooperative-sticky`: This strategy uses incremental cooperative rebalancing to reduce partition movement during rebalances.
   For more information, refer to the Kafka CooperativeStickyAssignor documentation, refer to [CooperativeStickyAssignor][].
 
+Use `group_rebalance_strategies` to advertise more than one strategy to the group coordinator, in order of preference.
+It accepts the same values as `group_rebalance_strategy`:
+
+```alloy
+group_rebalance_strategies = ["cooperative-sticky", "range"]
+```
+
 {{< admonition type="note" >}}
-The upstream OpenTelemetry Collector setting behind `group_rebalance_strategy` is deprecated in favor of an ordered list of strategies.
+The upstream OpenTelemetry Collector setting behind `group_rebalance_strategy` is deprecated in favor of `group_rebalance_strategies`.
 `group_rebalance_strategy` continues to work, and the `range` default is unchanged.
+The two arguments are mutually exclusive, so setting both fails to load.
 {{< /admonition >}}
 
 Using a `group_instance_id` is useful for stateful consumers or when you need to ensure that a specific consumer instance is always assigned the same set of partitions.
diff --git a/internal/component/otelcol/receiver/kafka/kafka.go b/internal/component/otelcol/receiver/kafka/kafka.go
index 846e3f33630..c05ebd00a2f 100644
--- a/internal/component/otelcol/receiver/kafka/kafka.go
+++ b/internal/component/otelcol/receiver/kafka/kafka.go
@@ -58,15 +58,16 @@ type Arguments struct {
 	HeaderExtraction HeaderExtraction                     `alloy:"header_extraction,block,optional"`
 	TLS              *otelcol.TLSClientArguments          `alloy:"tls,block,optional"`
 
-	MinFetchSize           int32         `alloy:"min_fetch_size,attr,optional"`
-	MaxFetchSize           int32         `alloy:"max_fetch_size,attr,optional"`
-	MaxPartitionFetchSize  int32         `alloy:"max_partition_fetch_size,attr,optional"`
-	MaxFetchWait           time.Duration `alloy:"max_fetch_wait,attr,optional"`
-	GroupRebalanceStrategy string        `alloy:"group_rebalance_strategy,attr,optional"`
-	GroupInstanceID        string        `alloy:"group_instance_id,attr,optional"`
-	RackID                 string        `alloy:"rack_id,attr,optional"`
-	UseLeaderEpoch         bool          `alloy:"use_leader_epoch,attr,optional"`
-	ConnIdleTimeout        time.Duration `alloy:"conn_idle_timeout,attr,optional"`
+	MinFetchSize             int32         `alloy:"min_fetch_size,attr,optional"`
+	MaxFetchSize             int32         `alloy:"max_fetch_size,attr,optional"`
+	MaxPartitionFetchSize    int32         `alloy:"max_partition_fetch_size,attr,optional"`
+	MaxFetchWait             time.Duration `alloy:"max_fetch_wait,attr,optional"`
+	GroupRebalanceStrategy   string        `alloy:"group_rebalance_strategy,attr,optional"`
+	GroupRebalanceStrategies []string      `alloy:"group_rebalance_strategies,attr,optional"`
+	GroupInstanceID          string        `alloy:"group_instance_id,attr,optional"`
+	RackID                   string        `alloy:"rack_id,attr,optional"`
+	UseLeaderEpoch           bool          `alloy:"use_leader_epoch,attr,optional"`
+	ConnIdleTimeout          time.Duration `alloy:"conn_idle_timeout,attr,optional"`
 
 	ErrorBackOff ErrorBackOffArguments `alloy:"error_backoff,block,optional"`
 
@@ -131,13 +132,34 @@ func (args *Arguments) Validate() error {
 		}
 	}
 
-	switch args.GroupRebalanceStrategy {
+	// Upstream rejects setting both forms. group_rebalance_strategy keeps its "range"
+	// default, so only treat a non-default value as an explicit conflict.
+	if len(args.GroupRebalanceStrategies) > 0 && args.GroupRebalanceStrategy != defaultGroupRebalanceStrategy {
+		return fmt.Errorf("group_rebalance_strategy and group_rebalance_strategies are mutually exclusive; group_rebalance_strategy is deprecated, prefer group_rebalance_strategies")
+	}
+
+	if len(args.GroupRebalanceStrategies) > 0 {
+		for _, strategy := range args.GroupRebalanceStrategies {
+			if err := validateGroupRebalanceStrategy(strategy); err != nil {
+				return err
+			}
+		}
+	} else if err := validateGroupRebalanceStrategy(args.GroupRebalanceStrategy); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+const defaultGroupRebalanceStrategy = "range"
+
+func validateGroupRebalanceStrategy(strategy string) error {
+	switch strategy {
 	case "range", "roundrobin", "sticky", "cooperative-sticky":
+		return nil
 	default:
 		return fmt.Errorf("group_rebalance_strategy must be one of 'range', 'roundrobin', 'sticky', or 'cooperative-sticky'")
 	}
-
-	return nil
 }
 
 type KafkaReceiverTopicEncodingConfig struct {
@@ -232,7 +254,17 @@ func (args Arguments) Convert() (otelcomponent.Config, error) {
 	result.ConsumerConfig.MaxFetchSize = args.MaxFetchSize
 	result.ConsumerConfig.MaxPartitionFetchSize = args.MaxPartitionFetchSize
 	result.ConsumerConfig.MaxFetchWait = args.MaxFetchWait
-	result.ConsumerConfig.GroupRebalanceStrategy = configkafka.GroupRebalanceStrategy(args.GroupRebalanceStrategy)
+	// Upstream rejects both forms being set, so send only the one in use.
+	if len(args.GroupRebalanceStrategies) > 0 {
+		strategies := make([]configkafka.GroupRebalanceStrategy, 0, len(args.GroupRebalanceStrategies))
+		for _, strategy := range args.GroupRebalanceStrategies {
+			strategies = append(strategies, configkafka.GroupRebalanceStrategy(strategy))
+		}
+		result.ConsumerConfig.GroupRebalanceStrategies = strategies
+		result.ConsumerConfig.GroupRebalanceStrategy = ""
+	} else {
+		result.ConsumerConfig.GroupRebalanceStrategy = configkafka.GroupRebalanceStrategy(args.GroupRebalanceStrategy)
+	}
 	result.ConsumerConfig.GroupInstanceID = args.GroupInstanceID
 	result.ClientConfig.RackID = args.RackID
 	result.ClientConfig.UseLeaderEpoch = args.UseLeaderEpoch
diff --git a/internal/component/otelcol/receiver/kafka/kafka_test.go b/internal/component/otelcol/receiver/kafka/kafka_test.go
index b821b5b522f..150efd63f35 100644
--- a/internal/component/otelcol/receiver/kafka/kafka_test.go
+++ b/internal/component/otelcol/receiver/kafka/kafka_test.go
@@ -699,3 +699,68 @@ func TestDebugMetricsConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestGroupRebalanceStrategies(t *testing.T) {
+	base := `
+		brokers          = ["broker:9092"]
+		protocol_version = "2.0.0"
+
+		output {}
+	`
+
+	convert := func(t *testing.T, cfg string) *kafkareceiver.Config {
+		var args kafka.Arguments
+		require.NoError(t, syntax.Unmarshal([]byte(cfg), &args))
+		converted, err := args.Convert()
+		require.NoError(t, err)
+		return converted.(*kafkareceiver.Config)
+	}
+
+	t.Run("defaults to the singular strategy", func(t *testing.T) {
+		otelObj := convert(t, base)
+
+		require.Equal(t, configkafka.GroupRebalanceStrategy("range"), otelObj.ConsumerConfig.GroupRebalanceStrategy)
+		require.Empty(t, otelObj.ConsumerConfig.GroupRebalanceStrategies)
+	})
+
+	t.Run("the plural form replaces the singular", func(t *testing.T) {
+		otelObj := convert(t, `
+			brokers                    = ["broker:9092"]
+			protocol_version           = "2.0.0"
+			group_rebalance_strategies = ["cooperative-sticky", "range"]
+
+			output {}
+		`)
+
+		require.Equal(t, []configkafka.GroupRebalanceStrategy{"cooperative-sticky", "range"}, otelObj.ConsumerConfig.GroupRebalanceStrategies)
+		// Upstream rejects both forms being set, so the singular must be cleared.
+		require.Empty(t, otelObj.ConsumerConfig.GroupRebalanceStrategy)
+	})
+
+	t.Run("setting both forms is rejected", func(t *testing.T) {
+		var args kafka.Arguments
+		err := syntax.Unmarshal([]byte(`
+			brokers                    = ["broker:9092"]
+			protocol_version           = "2.0.0"
+			group_rebalance_strategy   = "sticky"
+			group_rebalance_strategies = ["range"]
+
+			output {}
+		`), &args)
+
+		require.ErrorContains(t, err, "mutually exclusive")
+	})
+
+	t.Run("an invalid strategy in the list is rejected", func(t *testing.T) {
+		var args kafka.Arguments
+		err := syntax.Unmarshal([]byte(`
+			brokers                    = ["broker:9092"]
+			protocol_version           = "2.0.0"
+			group_rebalance_strategies = ["range", "nonsense"]
+
+			output {}
+		`), &args)
+
+		require.ErrorContains(t, err, "must be one of")
+	})
+}

From f85e81559de88e4356ad57ea1e76d0be1613bdf8 Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Thu, 13 Aug 2026 10:09:48 +0200
Subject: [PATCH 4/8] Update
 docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md

Co-authored-by: Clayton Cornell <131809008+clayton-cornell@users.noreply.github.com>
---
 .../reference/components/otelcol/otelcol.exporter.kafka.md      | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md b/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
index 74bb7d262ac..100a09a197c 100644
--- a/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
+++ b/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
@@ -226,7 +226,7 @@ The following arguments are supported:
 
 Refer to the [Go sarama documentation][RequiredAcks] for more information on `required_acks`.
 
-`max_broker_write_bytes` must be at least `104857600` (100 MiB), and `max_message_bytes` must be less than or equal to it.
+`max_broker_write_bytes` must be at least `104857600` (100 MiB), and `max_message_bytes` must be less than or equal to `max_broker_write_bytes`.
 Raise `max_broker_write_bytes` if you need a `max_message_bytes` larger than the default.
 
 Set `linger` to `"0s"` to send records as soon as they arrive, at the cost of less effective batching.

From dc597e3c81aad5b3a3d8b23cb7687f052f9e3b76 Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Thu, 13 Aug 2026 10:28:46 +0200
Subject: [PATCH 5/8] fix kafka exporter compression definition

---
 .../components/otelcol/otelcol.exporter.kafka.md       | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md b/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
index 100a09a197c..2c6199adc73 100644
--- a/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
+++ b/docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md
@@ -217,14 +217,14 @@ The following arguments are supported:
 
 | Name                     | Type       | Description                                                                | Default     | Required |
 | ------------------------ | ---------- | -------------------------------------------------------------------------- | ----------- | -------- |
-| `compression`            | `string`   | The level of compression to use on messages.                               | `"none"`    | no       |
+| `compression`            | `string`   | The compression algorithm to use on messages.                              | `"none"`    | no       |
 | `flush_max_messages`     | `number`   | The maximum number of messages in one request.                             | `10000`     | no       |
 | `linger`                 | `duration` | How long a topic partition waits for more records before building a request. | `"10ms"`    | no       |
 | `max_broker_write_bytes` | `number`   | The maximum permitted size of a single write to a broker in bytes.         | `104857600` | no       |
 | `max_message_bytes`      | `number`   | The maximum permitted size of a message in bytes.                          | `1000000`   | no       |
 | `required_acks`          | `number`   | Controls when a message is regarded as transmitted.                        | `1`         | no       |
 
-Refer to the [Go sarama documentation][RequiredAcks] for more information on `required_acks`.
+Refer to the [Kafka producer configuration documentation][RequiredAcks] for more information on `required_acks`.
 
 `max_broker_write_bytes` must be at least `104857600` (100 MiB), and `max_message_bytes` must be less than or equal to `max_broker_write_bytes`.
 Raise `max_broker_write_bytes` if you need a `max_message_bytes` larger than the default.
@@ -232,10 +232,10 @@ Raise `max_broker_write_bytes` if you need a `max_message_bytes` larger than the
 Set `linger` to `"0s"` to send records as soon as they arrive, at the cost of less effective batching.
 
 `compression` could be set to either `none`, `gzip`, `snappy`, `lz4`, or `zstd`.
-Refer to the [Go sarama documentation][CompressionCodec] for more information.
+Refer to the [franz-go documentation][CompressionCodec] for more information.
 
-[RequiredAcks]: https://pkg.go.dev/github.com/IBM/sarama@v1.43.2#RequiredAcks
-[CompressionCodec]: https://pkg.go.dev/github.com/IBM/sarama@v1.43.2#CompressionCodec
+[RequiredAcks]: https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks
+[CompressionCodec]: https://pkg.go.dev/github.com/twmb/franz-go/pkg/kgo#CompressionCodec
 
 ### `compression_params`
 

From c817f68d262d3f0f8273220fe491d85833a77792 Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Thu, 13 Aug 2026 10:39:28 +0200
Subject: [PATCH 6/8] document mutual exclusivity between
 group_relebance_strategy/ies

---
 .../reference/components/otelcol/otelcol.receiver.kafka.md    | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md b/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md
index 748e77c93c4..0a0839de77b 100644
--- a/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md
+++ b/docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md
@@ -51,8 +51,8 @@ You can use the following arguments with `otelcol.receiver.kafka`:
 | `encoding`                                 | `string`       | (Deprecated) Encoding of payload read from Kafka.                                                                     | `"otlp_proto"`     | no       |
 | `group_id`                                 | `string`       | Consumer group to consume messages from.                                                                              | `"otel-collector"` | no       |
 | `group_instance_id`                        | `string`       | A unique identifier for the consumer instance within a consumer group.                                                | `""`               | no       |
-| `group_rebalance_strategy`                 | `string`       | The strategy used to assign partitions to consumers within a consumer group.                                          | `"range"`          | no       |
-| `group_rebalance_strategies`               | `list(string)` | The ordered list of strategies to advertise to the group coordinator.                                                 | `[]`               | no       |
+| `group_rebalance_strategy`                 | `string`       | (Deprecated: use `group_rebalance_strategies` instead) The strategy used to assign partitions to consumers within a consumer group. Mutually exclusive with `group_rebalance_strategies`. | `"range"`          | no       |
+| `group_rebalance_strategies`               | `list(string)` | The ordered list of strategies to advertise to the group coordinator. Mutually exclusive with `group_rebalance_strategy`. | `[]`               | no       |
 | `heartbeat_interval`                       | `duration`     | The expected time between heartbeats to the consumer coordinator when using Kafka group management.                   | `"3s"`             | no       |
 | `initial_offset`                           | `string`       | Initial offset to use if no offset was previously committed.                                                          | `"latest"`         | no       |
 | `max_fetch_size`                           | `int`          | The maximum number of message bytes to fetch in a request.                                                            | `1048576`          | no       |

From b011e150ab8c4d6e00e9235e86c6067c7d66c6fa Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Thu, 13 Aug 2026 13:16:19 +0200
Subject: [PATCH 7/8] Ensure that empty group_rebalance_strategy or
 group_rebalance_strategies is handled correctly

---
 .../component/otelcol/receiver/kafka/kafka.go | 53 ++++++++++---------
 .../otelcolconvert/converter_kafkareceiver.go | 33 +++++++++---
 .../testdata/kafka_defaults.alloy             | 37 +++++++++++++
 .../testdata/kafka_defaults.yaml              | 20 +++++++
 .../testdata/kafka_strategies.alloy           | 37 +++++++++++++
 .../testdata/kafka_strategies.yaml            | 23 ++++++++
 6 files changed, 170 insertions(+), 33 deletions(-)
 create mode 100644 internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy
 create mode 100644 internal/converter/internal/otelcolconvert/testdata/kafka_defaults.yaml
 create mode 100644 internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy
 create mode 100644 internal/converter/internal/otelcolconvert/testdata/kafka_strategies.yaml

diff --git a/internal/component/otelcol/receiver/kafka/kafka.go b/internal/component/otelcol/receiver/kafka/kafka.go
index c05ebd00a2f..4c7142b55ea 100644
--- a/internal/component/otelcol/receiver/kafka/kafka.go
+++ b/internal/component/otelcol/receiver/kafka/kafka.go
@@ -86,20 +86,19 @@ func (args *Arguments) SetToDefault() {
 		// We use the defaults from the upstream OpenTelemetry Collector component
 		// for compatibility, even though that means using a client and group ID of
 		// "otel-collector".
-		Brokers:                []string{"localhost:9092"},
-		ClientID:               "otel-collector",
-		GroupID:                "otel-collector",
-		InitialOffset:          "latest",
-		SessionTimeout:         10 * time.Second,
-		HeartbeatInterval:      3 * time.Second,
-		MinFetchSize:           1,
-		MaxFetchSize:           1048576,
-		MaxPartitionFetchSize:  1048576,
-		MaxFetchWait:           250 * time.Millisecond,
-		GroupRebalanceStrategy: "range",
-		RackID:                 "",
-		UseLeaderEpoch:         true,
-		ConnIdleTimeout:        9 * time.Minute,
+		Brokers:               []string{"localhost:9092"},
+		ClientID:              "otel-collector",
+		GroupID:               "otel-collector",
+		InitialOffset:         "latest",
+		SessionTimeout:        10 * time.Second,
+		HeartbeatInterval:     3 * time.Second,
+		MinFetchSize:          1,
+		MaxFetchSize:          1048576,
+		MaxPartitionFetchSize: 1048576,
+		MaxFetchWait:          250 * time.Millisecond,
+		RackID:                "",
+		UseLeaderEpoch:        true,
+		ConnIdleTimeout:       9 * time.Minute,
 		Logs: KafkaReceiverTopicEncodingConfig{
 			Topics:   []string{"otlp_logs"},
 			Encoding: "otlp_proto",
@@ -132,20 +131,22 @@ func (args *Arguments) Validate() error {
 		}
 	}
 
-	// Upstream rejects setting both forms. group_rebalance_strategy keeps its "range"
-	// default, so only treat a non-default value as an explicit conflict.
-	if len(args.GroupRebalanceStrategies) > 0 && args.GroupRebalanceStrategy != defaultGroupRebalanceStrategy {
+	// Upstream rejects setting both forms, whatever their values.
+	if len(args.GroupRebalanceStrategies) > 0 && args.GroupRebalanceStrategy != "" {
 		return fmt.Errorf("group_rebalance_strategy and group_rebalance_strategies are mutually exclusive; group_rebalance_strategy is deprecated, prefer group_rebalance_strategies")
 	}
 
-	if len(args.GroupRebalanceStrategies) > 0 {
-		for _, strategy := range args.GroupRebalanceStrategies {
-			if err := validateGroupRebalanceStrategy(strategy); err != nil {
-				return err
-			}
+	for _, strategy := range args.GroupRebalanceStrategies {
+		if err := validateGroupRebalanceStrategy(strategy); err != nil {
+			return err
+		}
+	}
+
+	// An empty singular means unset; Convert applies the default.
+	if args.GroupRebalanceStrategy != "" {
+		if err := validateGroupRebalanceStrategy(args.GroupRebalanceStrategy); err != nil {
+			return err
 		}
-	} else if err := validateGroupRebalanceStrategy(args.GroupRebalanceStrategy); err != nil {
-		return err
 	}
 
 	return nil
@@ -262,8 +263,10 @@ func (args Arguments) Convert() (otelcomponent.Config, error) {
 		}
 		result.ConsumerConfig.GroupRebalanceStrategies = strategies
 		result.ConsumerConfig.GroupRebalanceStrategy = ""
-	} else {
+	} else if args.GroupRebalanceStrategy != "" {
 		result.ConsumerConfig.GroupRebalanceStrategy = configkafka.GroupRebalanceStrategy(args.GroupRebalanceStrategy)
+	} else {
+		result.ConsumerConfig.GroupRebalanceStrategy = defaultGroupRebalanceStrategy
 	}
 	result.ConsumerConfig.GroupInstanceID = args.GroupInstanceID
 	result.ClientConfig.RackID = args.RackID
diff --git a/internal/converter/internal/otelcolconvert/converter_kafkareceiver.go b/internal/converter/internal/otelcolconvert/converter_kafkareceiver.go
index cf7f3caebf4..6ea0d2a755d 100644
--- a/internal/converter/internal/otelcolconvert/converter_kafkareceiver.go
+++ b/internal/converter/internal/otelcolconvert/converter_kafkareceiver.go
@@ -58,6 +58,7 @@ func toKafkaReceiver(state *State, id componentstatus.InstanceID, cfg *kafkarece
 		tlsCfgPtr = &tlsCfg
 	}
 
+	rebalanceStrategy, rebalanceStrategies := toKafkaRebalance(cfg.ConsumerConfig)
 	return &kafka.Arguments{
 		Brokers:           cfg.ClientConfig.Brokers,
 		ProtocolVersion:   cfg.ClientConfig.ProtocolVersion,
@@ -82,14 +83,15 @@ func toKafkaReceiver(state *State, id componentstatus.InstanceID, cfg *kafkarece
 		Metrics: toKafkaTopicEncodingConfig(cfg.Metrics),
 		Traces:  toKafkaTopicEncodingConfig(cfg.Traces),
 
-		MinFetchSize:           cfg.ConsumerConfig.MinFetchSize,
-		MaxFetchSize:           cfg.ConsumerConfig.MaxFetchSize,
-		MaxPartitionFetchSize:  cfg.ConsumerConfig.MaxPartitionFetchSize,
-		MaxFetchWait:           cfg.ConsumerConfig.MaxFetchWait,
-		RackID:                 cfg.ClientConfig.RackID,
-		UseLeaderEpoch:         cfg.ClientConfig.UseLeaderEpoch,
-		GroupRebalanceStrategy: string(cfg.ConsumerConfig.GroupRebalanceStrategy),
-		GroupInstanceID:        cfg.ConsumerConfig.GroupInstanceID,
+		MinFetchSize:             cfg.ConsumerConfig.MinFetchSize,
+		MaxFetchSize:             cfg.ConsumerConfig.MaxFetchSize,
+		MaxPartitionFetchSize:    cfg.ConsumerConfig.MaxPartitionFetchSize,
+		MaxFetchWait:             cfg.ConsumerConfig.MaxFetchWait,
+		RackID:                   cfg.ClientConfig.RackID,
+		UseLeaderEpoch:           cfg.ClientConfig.UseLeaderEpoch,
+		GroupRebalanceStrategy:   rebalanceStrategy,
+		GroupRebalanceStrategies: rebalanceStrategies,
+		GroupInstanceID:          cfg.ConsumerConfig.GroupInstanceID,
 
 		ErrorBackOff: toKafkaErrorBackOff(cfg.ErrorBackOff),
 
@@ -241,3 +243,18 @@ func toKafkaHeaderExtraction(cfg kafkareceiver.HeaderExtraction) kafka.HeaderExt
 		Headers:        cfg.Headers,
 	}
 }
+
+func toKafkaRebalance(cfg configkafka.ConsumerConfig) (strategy string, strategies []string) {
+	switch {
+	case len(cfg.GroupRebalanceStrategies) > 0:
+		strategies = make([]string, 0, len(cfg.GroupRebalanceStrategies))
+		for _, s := range cfg.GroupRebalanceStrategies {
+			strategies = append(strategies, string(s))
+		}
+		return "", strategies
+	case cfg.GroupRebalanceStrategy != "":
+		return string(cfg.GroupRebalanceStrategy), nil
+	default:
+		return string(configkafka.CooperativeStickyBalanceStrategy), nil
+	}
+}
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy b/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy
new file mode 100644
index 00000000000..47c0c38a31c
--- /dev/null
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy
@@ -0,0 +1,37 @@
+otelcol.receiver.kafka "default" {
+	brokers                  = ["broker:9092"]
+	protocol_version         = "2.0.0"
+	group_rebalance_strategy = "cooperative-sticky"
+
+	output {
+		logs = [otelcol.exporter.kafka.default.input]
+	}
+}
+
+otelcol.exporter.kafka "default" {
+	protocol_version = "2.0.0"
+	brokers          = ["redpanda:9092"]
+
+	logs {
+		topic    = "otlp_logs"
+		encoding = "otlp_proto"
+	}
+
+	metrics {
+		topic    = "otlp_metrics"
+		encoding = "otlp_proto"
+	}
+
+	traces {
+		topic    = "otlp_spans"
+		encoding = "otlp_proto"
+	}
+
+	producer {
+		max_message_bytes         = 1000000
+		required_acks             = 1
+		compression               = "none"
+		flush_max_messages        = 10000
+		allow_auto_topic_creation = true
+	}
+}
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.yaml b/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.yaml
new file mode 100644
index 00000000000..11fc36942e0
--- /dev/null
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.yaml
@@ -0,0 +1,20 @@
+receivers:
+  kafka:
+    brokers:
+      - broker:9092
+    protocol_version: 2.0.0
+    # group_rebalance_strategy is deliberately unset. Upstream leaves it empty and
+    # lets franz-go pick cooperative-sticky, so the converter must emit that
+    # explicitly rather than fall back to Alloy's differing "range" default.
+
+exporters:
+  kafka:
+    brokers:
+      - redpanda:9092
+    protocol_version: 2.0.0
+
+service:
+  pipelines:
+    logs:
+      receivers: [kafka]
+      exporters: [kafka]
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy b/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy
new file mode 100644
index 00000000000..37ab8aa9dc9
--- /dev/null
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy
@@ -0,0 +1,37 @@
+otelcol.receiver.kafka "default" {
+	brokers                    = ["broker:9092"]
+	protocol_version           = "2.0.0"
+	group_rebalance_strategies = ["cooperative-sticky", "range"]
+
+	output {
+		logs = [otelcol.exporter.kafka.default.input]
+	}
+}
+
+otelcol.exporter.kafka "default" {
+	protocol_version = "2.0.0"
+	brokers          = ["redpanda:9092"]
+
+	logs {
+		topic    = "otlp_logs"
+		encoding = "otlp_proto"
+	}
+
+	metrics {
+		topic    = "otlp_metrics"
+		encoding = "otlp_proto"
+	}
+
+	traces {
+		topic    = "otlp_spans"
+		encoding = "otlp_proto"
+	}
+
+	producer {
+		max_message_bytes         = 1000000
+		required_acks             = 1
+		compression               = "none"
+		flush_max_messages        = 10000
+		allow_auto_topic_creation = true
+	}
+}
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.yaml b/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.yaml
new file mode 100644
index 00000000000..a5afdda10a9
--- /dev/null
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.yaml
@@ -0,0 +1,23 @@
+receivers:
+  kafka:
+    brokers:
+      - broker:9092
+    protocol_version: 2.0.0
+    # The plural form supersedes the deprecated group_rebalance_strategy. The
+    # converter must carry the ordered list across and leave the singular unset,
+    # since upstream rejects both being set.
+    group_rebalance_strategies:
+      - cooperative-sticky
+      - range
+
+exporters:
+  kafka:
+    brokers:
+      - redpanda:9092
+    protocol_version: 2.0.0
+
+service:
+  pipelines:
+    logs:
+      receivers: [kafka]
+      exporters: [kafka]

From a5c62fe9f3e4b5e797947d0a8b9c758653519eb4 Mon Sep 17 00:00:00 2001
From: Bejal Lewis <164711649+blewis12@users.noreply.github.com>
Date: Thu, 13 Aug 2026 13:44:37 +0200
Subject: [PATCH 8/8] include MaxBrokerWriteBytes and Linger in conversion

---
 .../internal/otelcolconvert/converter_kafkaexporter.go    | 2 ++
 .../internal/otelcolconvert/testdata/kafka.alloy          | 6 ++++--
 .../converter/internal/otelcolconvert/testdata/kafka.yaml | 4 +++-
 .../internal/otelcolconvert/testdata/kafka_defaults.alloy | 8 --------
 .../otelcolconvert/testdata/kafka_strategies.alloy        | 8 --------
 5 files changed, 9 insertions(+), 19 deletions(-)

diff --git a/internal/converter/internal/otelcolconvert/converter_kafkaexporter.go b/internal/converter/internal/otelcolconvert/converter_kafkaexporter.go
index c14429a9b65..d197574e3e5 100644
--- a/internal/converter/internal/otelcolconvert/converter_kafkaexporter.go
+++ b/internal/converter/internal/otelcolconvert/converter_kafkaexporter.go
@@ -94,11 +94,13 @@ func toKafkaExporter(cfg *kafkaexporter.Config) *kafka.Arguments {
 func toKafkaProducer(cfg configkafka.ProducerConfig) kafka.Producer {
 	return kafka.Producer{
 		MaxMessageBytes:        cfg.MaxMessageBytes,
+		MaxBrokerWriteBytes:    cfg.MaxBrokerWriteBytes,
 		Compression:            cfg.Compression,
 		CompressionParams:      toKafkaCompressionParams(cfg.CompressionParams),
 		RequiredAcks:           int(cfg.RequiredAcks),
 		FlushMaxMessages:       cfg.FlushMaxMessages,
 		AllowAutoTopicCreation: cfg.AllowAutoTopicCreation,
+		Linger:                 cfg.Linger,
 	}
 }
 
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka.alloy b/internal/converter/internal/otelcolconvert/testdata/kafka.alloy
index 4da66bc0ae6..30a0affcadf 100644
--- a/internal/converter/internal/otelcolconvert/testdata/kafka.alloy
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka.alloy
@@ -115,13 +115,15 @@ otelcol.exporter.kafka "default" {
 	}
 
 	producer {
-		max_message_bytes = 1000001
-		compression       = "gzip"
+		max_message_bytes      = 209715200
+		max_broker_write_bytes = 209715200
+		compression            = "gzip"
 
 		compression_params {
 			level = 9
 		}
 		flush_max_messages        = 11
 		allow_auto_topic_creation = true
+		linger                    = "250ms"
 	}
 }
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka.yaml b/internal/converter/internal/otelcolconvert/testdata/kafka.yaml
index e24cb84515a..c08f52c55b8 100644
--- a/internal/converter/internal/otelcolconvert/testdata/kafka.yaml
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka.yaml
@@ -71,7 +71,9 @@ exporters:
         service_name: someservice
         realm: myrealm
     producer:
-      max_message_bytes: 1000001
+      max_message_bytes: 209715200
+      max_broker_write_bytes: 209715200
+      linger: 250ms
       compression: gzip
       required_acks: 0
       flush_max_messages: 11
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy b/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy
index 47c0c38a31c..3e36000e957 100644
--- a/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka_defaults.alloy
@@ -26,12 +26,4 @@ otelcol.exporter.kafka "default" {
 		topic    = "otlp_spans"
 		encoding = "otlp_proto"
 	}
-
-	producer {
-		max_message_bytes         = 1000000
-		required_acks             = 1
-		compression               = "none"
-		flush_max_messages        = 10000
-		allow_auto_topic_creation = true
-	}
 }
diff --git a/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy b/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy
index 37ab8aa9dc9..e2db93d4530 100644
--- a/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy
+++ b/internal/converter/internal/otelcolconvert/testdata/kafka_strategies.alloy
@@ -26,12 +26,4 @@ otelcol.exporter.kafka "default" {
 		topic    = "otlp_spans"
 		encoding = "otlp_proto"
 	}
-
-	producer {
-		max_message_bytes         = 1000000
-		required_acks             = 1
-		compression               = "none"
-		flush_max_messages        = 10000
-		allow_auto_topic_creation = true
-	}
 }