[exporter/kafka] add compression level to kafka configuration (#39647)

#### Description

Add support for configuring Kafka producer compression level.

Both Kafka and the Sarama library support setting compression level, but
the OpenTelemetry Collector currently does not expose this option.
This change adds a `compression_level` configuration field to the Kafka
exporter, allowing users to control compression level explicitly.

#### Link to tracking issue

_No tracking issue provided._

#### Testing

- Added unit tests

#### Documentation

- Updated configuration documentation alongside the existing
`compression` field to describe the usage of `compression_level`.

---------

Signed-off-by: vidam.io <[email protected]>
Co-authored-by: Antoine Toulme <[email protected]>

Author: vidam-io

.chloggen/add-compression-level-in-kafka-producer.yaml

@@ -0,0 +1,27 @@
+# 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. filelogreceiver)
+component: kafkaexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add compression level in kafka producer.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39772]
+
+# (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:
+
+# 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

@@ -89,6 +89,22 @@ The following settings can be optionally configured:
   - `max_message_bytes` (default = 1000000) the maximum permitted size of a message in bytes
   - `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`
+    - `level` (default = -1) the compression level used when producing messages to kafka.
+    - The following are valid combinations of `compression` and `level`
+      - `gzip`
+        - BestSpeed: `1`
+        - BestCompression: `9`
+        - DefaultCompression: `-1`
+      - `zstd`
+        - SpeedFastest: `1`
+        - SpeedDefault: `3`
+        - SpeedBetterCompression: `6`
+        - SpeedBestCompression: `11`
+      - `lz4`
+        Only supports fast level
+      - `snappy`
+        No compression levels supported yet
   - `flush_max_messages` (default = 0) The maximum number of messages the producer will send in a single broker request.
 
 ### Supported encodings

exporter/kafkaexporter/go.mod

@@ -90,6 +90,7 @@ require (
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/config/configcompression v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/config/configtls v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/consumer/consumertest v0.125.1-0.20250505155216-829157cef7bb // indirect

exporter/kafkaexporter/go.sum

@@ -69,6 +69,7 @@ require (
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/config/configcompression v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/config/configtls v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/featuregate v1.31.1-0.20250505152726-56c7da210783 // indirect

extension/observer/kafkatopicsobserver/go.mod

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/IBM/sarama"
+	"go.opentelemetry.io/collector/config/configcompression"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka/configkafka"
 )
@@ -21,6 +22,13 @@ var saramaCompressionCodecs = map[string]sarama.CompressionCodec{
 	"zstd":   sarama.CompressionZSTD,
 }
 
+func convertToSaramaCompressionLevel(p configcompression.Level) int {
+	if p == configcompression.DefaultCompressionLevel {
+		return sarama.CompressionLevelDefault
+	}
+	return int(p)
+}
+
 var saramaInitialOffsets = map[string]int64{
 	configkafka.EarliestOffset: sarama.OffsetOldest,
 	configkafka.LatestOffset:   sarama.OffsetNewest,
@@ -96,6 +104,7 @@ func NewSaramaSyncProducer(
 	saramaConfig.Producer.RequiredAcks = sarama.RequiredAcks(producerConfig.RequiredAcks)
 	saramaConfig.Producer.Timeout = producerTimeout
 	saramaConfig.Producer.Compression = saramaCompressionCodecs[producerConfig.Compression]
+	saramaConfig.Producer.CompressionLevel = convertToSaramaCompressionLevel(producerConfig.CompressionParams.Level)
 	return sarama.NewSyncProducer(clientConfig.Brokers, saramaConfig)
 }
 

extension/observer/kafkatopicsobserver/go.sum

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/IBM/sarama"
+	"go.opentelemetry.io/collector/config/configcompression"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/confmap"
 )
@@ -184,6 +185,9 @@ type ProducerConfig struct {
 	// The options are: 'none' (default), 'gzip', 'snappy', 'lz4', and 'zstd'
 	Compression string `mapstructure:"compression"`
 
+	// CompressionParams defines compression parameters for the producer.
+	CompressionParams configcompression.CompressionParams `mapstructure:"compression_params"`
+
 	// The maximum number of messages the producer will send in a single
 	// broker request. Defaults to 0 for unlimited. Similar to
 	// `queue.buffering.max.messages` in the JVM producer.
@@ -192,17 +196,26 @@ type ProducerConfig struct {
 
 func NewDefaultProducerConfig() ProducerConfig {
 	return ProducerConfig{
-		MaxMessageBytes:  1000000,
-		RequiredAcks:     WaitForLocal,
-		Compression:      "none",
+		MaxMessageBytes: 1000000,
+		RequiredAcks:    WaitForLocal,
+		Compression:     "none",
+		CompressionParams: configcompression.CompressionParams{
+			Level: configcompression.DefaultCompressionLevel,
+		},
 		FlushMaxMessages: 0,
 	}
 }
 
 func (c ProducerConfig) Validate() error {
 	switch c.Compression {
 	case "none", "gzip", "snappy", "lz4", "zstd":
-		// Valid compression
+		ct := configcompression.Type(c.Compression)
+		if !ct.IsCompressed() {
+			return nil
+		}
+		if err := ct.ValidateParams(c.CompressionParams); err != nil {
+			return err
+		}
 	default:
 		return fmt.Errorf(
 			"compression should be one of 'none', 'gzip', 'snappy', 'lz4', or 'zstd'. configured value is %q",

internal/kafka/client.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configcompression"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
 	"go.opentelemetry.io/collector/confmap/xconfmap"
@@ -162,12 +163,29 @@ func TestProducerConfig(t *testing.T) {
 		},
 		"full": {
 			expected: ProducerConfig{
-				MaxMessageBytes:  1,
-				RequiredAcks:     0,
-				Compression:      "gzip",
+				MaxMessageBytes: 1,
+				RequiredAcks:    0,
+				Compression:     "gzip",
+				CompressionParams: configcompression.CompressionParams{
+					Level: 1,
+				},
 				FlushMaxMessages: 2,
 			},
 		},
+		"default_compression_level": {
+			expected: ProducerConfig{
+				MaxMessageBytes: 1,
+				RequiredAcks:    0,
+				Compression:     "zstd",
+				CompressionParams: configcompression.CompressionParams{
+					Level: configcompression.DefaultCompressionLevel,
+				},
+				FlushMaxMessages: 2,
+			},
+		},
+		"invalid_compression_level": {
+			expectedErr: `unsupported parameters {Level:-123} for compression type "gzip"`,
+		},
 		"required_acks_all": {
 			expected: func() ProducerConfig {
 				cfg := NewDefaultProducerConfig()

internal/kafka/configkafka/config.go

@@ -3,6 +3,20 @@ kafka/full:
   max_message_bytes: 1
   required_acks: 0
   compression: gzip
+  compression_params:
+    level: 1
+  flush_max_messages: 2
+kafka/default_compression_level:
+  max_message_bytes: 1
+  required_acks: 0
+  compression: zstd
+  flush_max_messages: 2
+kafka/invalid_compression_level:
+  max_message_bytes: 1
+  required_acks: 0
+  compression: gzip
+  compression_params:
+    level: -123
   flush_max_messages: 2
 kafka/required_acks_all:
   required_acks: all

internal/kafka/configkafka/config_test.go

@@ -12,6 +12,7 @@ require (
 	github.com/twmb/franz-go/pkg/kfake v0.0.0-20250320172111-35ab5e5f5327
 	github.com/xdg-go/scram v1.1.2
 	go.opentelemetry.io/collector/component v1.31.1-0.20250505152726-56c7da210783
+	go.opentelemetry.io/collector/config/configcompression v1.31.1-0.20250505152726-56c7da210783
 	go.opentelemetry.io/collector/config/configopaque v1.31.1-0.20250505152726-56c7da210783
 	go.opentelemetry.io/collector/config/configtls v1.31.1-0.20250505152726-56c7da210783
 	go.opentelemetry.io/collector/confmap v1.31.1-0.20250505152726-56c7da210783

internal/kafka/configkafka/testdata/producer_config.yaml

@@ -78,6 +78,7 @@ require (
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/config/configcompression v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/config/configtls v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.125.1-0.20250505155216-829157cef7bb // indirect

internal/kafka/go.mod

@@ -101,6 +101,7 @@ require (
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/config/configcompression v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.125.1-0.20250505155216-829157cef7bb // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.125.1-0.20250505155216-829157cef7bb // indirect