fix(kafkaexporter): Make Sarama's ConfigurationError as permanent to prevent retries (open-telemetry#38608)

#### Description
This fix unifies message send error handling for all types of telemetry.
It is designed to identify whether the error was caused by a
ConfigurationError and then reclassify it as a permanent consumer error
to prevent further retries.

#### Link to tracking issue

open-telemetry#38604

#### Testing
Unit test coverage added

#### Documentation
No changes

Co-authored-by: Antoine Toulme <[email protected]>
Co-authored-by: Sean Marciniak <[email protected]>

Author: 3 people

.chloggen/fix_kafkaexporter_permanent_config_error.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: kafkaexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Make Sarama's ConfigurationError as permanent to prevent retries"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38604]
+
+# (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/kafka_exporter.go

@@ -117,13 +117,7 @@ func (e *kafkaExporter[T]) exportData(ctx context.Context, data T) error {
 		ctx, e.cfg.IncludeMetadataKeys,
 	))
 	if err := e.producer.SendMessages(allSaramaMessages); err != nil {
-		var prodErr sarama.ProducerErrors
-		if errors.As(err, &prodErr) {
-			if len(prodErr) > 0 {
-				return kafkaErrors{len(prodErr), prodErr[0].Err.Error()}
-			}
-		}
-		return err
+		return wrapKafkaProducerError(err)
 	}
 	return nil
 }
@@ -342,3 +336,24 @@ func metadataToHeaders(ctx context.Context, keys []string) []sarama.RecordHeader
 	}
 	return headers
 }
+
+func wrapKafkaProducerError(err error) error {
+	var prodErr sarama.ProducerErrors
+	if !errors.As(err, &prodErr) || len(prodErr) == 0 {
+		return err
+	}
+
+	var areConfigErrs bool
+	var confErr sarama.ConfigurationError
+	for _, producerErr := range prodErr {
+		if areConfigErrs = errors.As(producerErr.Err, &confErr); !areConfigErrs {
+			break
+		}
+	}
+
+	if areConfigErrs {
+		return consumererror.NewPermanent(confErr)
+	}
+
+	return kafkaErrors{len(prodErr), prodErr[0].Err.Error()}
+}

exporter/kafkaexporter/kafka_exporter_test.go

@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -123,6 +124,27 @@ func TestTracesPusher_err(t *testing.T) {
 	assert.EqualError(t, err, expErr.Error())
 }
 
+func TestTracesPusher_conf_err(t *testing.T) {
+	t.Run("should return permanent err on config error", func(t *testing.T) {
+		expErr := sarama.ConfigurationError("configuration error")
+		prodErrs := sarama.ProducerErrors{
+			&sarama.ProducerError{Err: expErr},
+		}
+		host := extensionsHost{
+			component.MustNewID("trace_encoding"): ptraceMarshalerFuncExtension(func(ptrace.Traces) ([]byte, error) {
+				return nil, prodErrs
+			}),
+		}
+		config := createDefaultConfig().(*Config)
+		config.Traces.Encoding = "trace_encoding"
+		exp, _ := newMockTracesExporter(t, *config, host)
+
+		err := exp.exportData(context.Background(), testdata.GenerateTraces(2))
+
+		assert.True(t, consumererror.IsPermanent(err))
+	})
+}
+
 func TestTracesPusher_marshal_error(t *testing.T) {
 	marshalErr := errors.New("failed to marshal")
 	host := extensionsHost{
@@ -349,6 +371,27 @@ func TestMetricsPusher_err(t *testing.T) {
 	assert.EqualError(t, err, expErr.Error())
 }
 
+func TestMetricsPusher_conf_err(t *testing.T) {
+	t.Run("should return permanent err on config error", func(t *testing.T) {
+		expErr := sarama.ConfigurationError("configuration error")
+		prodErrs := sarama.ProducerErrors{
+			&sarama.ProducerError{Err: expErr},
+		}
+		host := extensionsHost{
+			component.MustNewID("metric_encoding"): ptraceMarshalerFuncExtension(func(ptrace.Traces) ([]byte, error) {
+				return nil, prodErrs
+			}),
+		}
+		config := createDefaultConfig().(*Config)
+		config.Traces.Encoding = "metric_encoding"
+		exp, _ := newMockTracesExporter(t, *config, host)
+
+		err := exp.exportData(context.Background(), testdata.GenerateTraces(2))
+
+		assert.True(t, consumererror.IsPermanent(err))
+	})
+}
+
 func TestMetricsPusher_marshal_error(t *testing.T) {
 	marshalErr := errors.New("failed to marshal")
 	host := extensionsHost{
@@ -521,6 +564,27 @@ func TestLogsPusher_err(t *testing.T) {
 	assert.EqualError(t, err, expErr.Error())
 }
 
+func TestLogsPusher_conf_err(t *testing.T) {
+	t.Run("should return permanent err on config error", func(t *testing.T) {
+		expErr := sarama.ConfigurationError("configuration error")
+		prodErrs := sarama.ProducerErrors{
+			&sarama.ProducerError{Err: expErr},
+		}
+		host := extensionsHost{
+			component.MustNewID("log_encoding"): ptraceMarshalerFuncExtension(func(ptrace.Traces) ([]byte, error) {
+				return nil, prodErrs
+			}),
+		}
+		config := createDefaultConfig().(*Config)
+		config.Traces.Encoding = "log_encoding"
+		exp, _ := newMockTracesExporter(t, *config, host)
+
+		err := exp.exportData(context.Background(), testdata.GenerateTraces(2))
+
+		assert.True(t, consumererror.IsPermanent(err))
+	})
+}
+
 func TestLogsPusher_marshal_error(t *testing.T) {
 	marshalErr := errors.New("failed to marshal")
 	host := extensionsHost{
@@ -808,3 +872,64 @@ func newMockLogsExporter(t *testing.T, cfg Config, host component.Host) (*kafkaE
 	})
 	return exp, producer
 }
+
+func TestWrapKafkaProducerError(t *testing.T) {
+	t.Run("should return permanent error on configuration error", func(t *testing.T) {
+		err := sarama.ConfigurationError("configuration error")
+		prodErrs := sarama.ProducerErrors{
+			&sarama.ProducerError{Err: err},
+		}
+
+		got := wrapKafkaProducerError(prodErrs)
+
+		assert.True(t, consumererror.IsPermanent(got))
+		assert.Contains(t, got.Error(), err.Error())
+	})
+
+	t.Run("should return permanent error whne multiple configuration error", func(t *testing.T) {
+		err := sarama.ConfigurationError("configuration error")
+		prodErrs := sarama.ProducerErrors{
+			&sarama.ProducerError{Err: err},
+			&sarama.ProducerError{Err: err},
+		}
+
+		got := wrapKafkaProducerError(prodErrs)
+
+		assert.True(t, consumererror.IsPermanent(got))
+		assert.Contains(t, got.Error(), err.Error())
+	})
+
+	t.Run("should return not permanent error when at least one not configuration error", func(t *testing.T) {
+		err := sarama.ConfigurationError("configuration error")
+		prodErrs := sarama.ProducerErrors{
+			&sarama.ProducerError{Err: err},
+			&sarama.ProducerError{Err: errors.New("other producer error")},
+		}
+
+		got := wrapKafkaProducerError(prodErrs)
+
+		assert.False(t, consumererror.IsPermanent(got))
+		assert.Contains(t, got.Error(), err.Error())
+	})
+
+	t.Run("should return not permanent error on other producer error", func(t *testing.T) {
+		err := errors.New("other producer error")
+		prodErrs := sarama.ProducerErrors{
+			&sarama.ProducerError{Err: err},
+		}
+
+		got := wrapKafkaProducerError(prodErrs)
+
+		assert.False(t, consumererror.IsPermanent(got))
+		assert.Contains(t, got.Error(), err.Error())
+	})
+
+	t.Run("should return not permanent error when other error", func(t *testing.T) {
+		err := errors.New("other error")
+
+		got := wrapKafkaProducerError(err)
+
+		assert.False(t, consumererror.IsPermanent(got))
+		assert.Contains(t, got.Error(), err.Error())
+	})
+}