feat: Expose new kafka receiver/exporter knobs after upgrade to v0.158.0

docs/sources/reference/components/otelcol/otelcol.exporter.kafka.md

@@ -215,20 +215,27 @@ 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 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.
+
+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`
 

docs/sources/reference/components/otelcol/otelcol.receiver.kafka.md

@@ -51,7 +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_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       |
@@ -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.

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"},

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)
+	})
+}

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"`
 
@@ -85,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",
@@ -131,13 +131,36 @@ func (args *Arguments) Validate() error {
 		}
 	}
 
-	switch args.GroupRebalanceStrategy {
+	// 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")
+	}
+
+	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
+		}
+	}
+
+	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 +255,19 @@ 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 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
 	result.ClientConfig.UseLeaderEpoch = args.UseLeaderEpoch