Merge branch 'main' into add-splunk-rolling-restart-metric

Author: juanhotencoding

.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