[exporter/awskinesisexporter] fix in compressor crashing under heady load due non-safe thread execution

.chloggen/fix_compressor-kinesis-exporter-thread-safe.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: 'bug_fix'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: awskinesisexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: the compressor was crashing under high load due it not being thread safe.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32589]
+
+# (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: Instead of using a mutex in the compressor Do I ended up creating a new compressor on each batch generation because it generated more performant execution in my tests with heavy load, specially if the payloads were bigger
+
+# 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: []

exporter/awskinesisexporter/exporter.go

@@ -19,7 +19,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/batch"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/compress"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/producer"
 )
 
@@ -90,16 +89,11 @@ func createExporter(ctx context.Context, c component.Config, log *zap.Logger, op
 		return nil, err
 	}
 
-	compressor, err := compress.NewCompressor(conf.Encoding.Compression)
-	if err != nil {
-		return nil, err
-	}
-
 	encoder, err := batch.NewEncoder(
 		conf.Encoding.Name,
 		batch.WithMaxRecordSize(conf.MaxRecordSize),
 		batch.WithMaxRecordsPerBatch(conf.MaxRecordsPerBatch),
-		batch.WithCompression(compressor),
+		batch.WithCompressionType(conf.Compression),
 	)
 
 	if err != nil {

exporter/awskinesisexporter/internal/batch/batch.go

@@ -5,12 +5,11 @@ package batch // import "github.com/open-telemetry/opentelemetry-collector-contr
 
 import (
 	"errors"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/compress"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/kinesis/types" //nolint:staticcheck // Some encoding types uses legacy prototype version
 	"go.opentelemetry.io/collector/consumer/consumererror"
-
-	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awskinesisexporter/internal/compress"
 )
 
 const (
@@ -29,7 +28,7 @@ type Batch struct {
 	maxBatchSize  int
 	maxRecordSize int
 
-	compression compress.Compressor
+	compressionType string
 
 	records []types.PutRecordsRequestEntry
 }
@@ -54,20 +53,18 @@ func WithMaxRecordSize(size int) Option {
 	}
 }
 
-func WithCompression(compressor compress.Compressor) Option {
+func WithCompressionType(compressionType string) Option {
 	return func(bt *Batch) {
-		if compressor != nil {
-			bt.compression = compressor
-		}
+		bt.compressionType = compressionType
 	}
 }
 
 func New(opts ...Option) *Batch {
 	bt := &Batch{
-		maxBatchSize:  MaxBatchedRecords,
-		maxRecordSize: MaxRecordSize,
-		compression:   compress.NewNoopCompressor(),
-		records:       make([]types.PutRecordsRequestEntry, 0, MaxBatchedRecords),
+		maxBatchSize:    MaxBatchedRecords,
+		maxRecordSize:   MaxRecordSize,
+		compressionType: "none",
+		records:         make([]types.PutRecordsRequestEntry, 0, MaxBatchedRecords),
 	}
 
 	for _, op := range opts {
@@ -78,7 +75,13 @@ func New(opts ...Option) *Batch {
 }
 
 func (b *Batch) AddRecord(raw []byte, key string) error {
-	record, err := b.compression.Do(raw)
+
+	compressor, err := compress.NewCompressor(b.compressionType)
+	if err != nil {
+		return err
+	}
+
+	record, err := compressor.Do(raw)
 	if err != nil {
 		return err
 	}

exporter/awskinesisexporter/internal/compress/noop_compression.go

@@ -9,12 +9,6 @@ type noop struct {
 	data io.Writer
 }
 
-func NewNoopCompressor() Compressor {
-	return &compressor{
-		compression: &noop{},
-	}
-}
-
 func (n *noop) Reset(w io.Writer) {
 	n.data = w
 }