Support for compression configs (#44)

Support for compression configs 
Co-authored-by: Apoorv Jain

exporter/kafkaexporter/compression.go

@@ -0,0 +1,43 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kafkaexporter
+
+import (
+	"strings"
+
+	"github.com/Shopify/sarama"
+)
+
+// Compression defines the compression method and the compression level.
+type Compression struct {
+	Codec string `mapstructure:"codec"`
+	Level int    `mapstructure:"level"`
+}
+
+func configureCompression(comp Compression, saramaConfig *sarama.Config) {
+	switch strings.ToLower(comp.Codec) {
+	case "none":
+		saramaConfig.Producer.Compression = sarama.CompressionNone
+	case "gzip":
+		saramaConfig.Producer.Compression = sarama.CompressionGZIP
+	case "snappy":
+		saramaConfig.Producer.Compression = sarama.CompressionSnappy
+	case "lz4":
+		saramaConfig.Producer.Compression = sarama.CompressionLZ4
+	case "zstd":
+		saramaConfig.Producer.Compression = sarama.CompressionZSTD
+	}
+	saramaConfig.Producer.CompressionLevel = comp.Level
+}

exporter/kafkaexporter/config.go

@@ -44,6 +44,9 @@ type Config struct {
 
 	// Authentication defines used authentication mechanism.
 	Authentication Authentication `mapstructure:"auth"`
+
+	// Compression defines the compression method and compression level, if applicable.
+	Compression Compression `mapstructure:"compression"`
 }
 
 // Metadata defines configuration for retrieving metadata from the broker.

exporter/kafkaexporter/config_test.go

@@ -74,5 +74,9 @@ func TestLoadConfig(t *testing.T) {
 				Backoff: defaultMetadataRetryBackoff,
 			},
 		},
+		Compression: Compression{
+			Codec: "gzip",
+			Level: 8,
+		},
 	}, c)
 }

exporter/kafkaexporter/kafka_exporter.go

@@ -88,6 +88,7 @@ func newSaramaProducer(config Config) (sarama.SyncProducer, error) {
 	c.Metadata.Full = config.Metadata.Full
 	c.Metadata.Retry.Max = config.Metadata.Retry.Max
 	c.Metadata.Retry.Backoff = config.Metadata.Retry.Backoff
+	configureCompression(config.Compression, c)
 	if config.ProtocolVersion != "" {
 		version, err := sarama.ParseKafkaVersion(config.ProtocolVersion)
 		if err != nil {

exporter/kafkaexporter/testdata/config.yaml

@@ -22,6 +22,9 @@ exporters:
       initial_interval: 10s
       max_interval: 60s
       max_elapsed_time: 10m
+    compression:
+      codec: gzip
+      level: 8
 
 processors:
   exampleprocessor: