fix(kafkareceiver): enforce a backoff mechanism on exporterhelper.ErrQueueIsFull error (#39581)

#### Description

In the current implementation, the BackOff mechanism is trigerred only
by [memory limiter
error](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/receiver/kafkareceiver/kafka_receiver.go#L562-L564).
However, the same behavior is expected for the
`exporterhelper.ErrQueueIsFull` error, which occurs when there is a
sending queue overflow (both [memory
queue](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/internal/queuebatch/memory_queue.go#L109-L112)
and [persistent
queue](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/internal/queuebatch/persistent_queue.go#L244-L247)).

#### Link to tracking issue
#39580

#### Testing
Unit coverage added

#### Documentation
No documentation updated

---------

Co-authored-by: Andrew Wilkins <[email protected]>
Co-authored-by: Sean Marciniak <[email protected]>

Author: 3 people

.chloggen/fix_kafkareceiver-backoff-on-queue-batch.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: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: enforce a backoff mechanism on non-permanent errors, such as when the queue is full
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39580]
+
+# (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]

receiver/kafkareceiver/go.mod

@@ -28,7 +28,9 @@ require (
 	go.opentelemetry.io/collector/confmap v1.33.0
 	go.opentelemetry.io/collector/confmap/xconfmap v0.127.0
 	go.opentelemetry.io/collector/consumer v1.33.0
+	go.opentelemetry.io/collector/consumer/consumererror v0.127.0
 	go.opentelemetry.io/collector/consumer/consumertest v0.127.0
+	go.opentelemetry.io/collector/exporter v0.127.0
 	go.opentelemetry.io/collector/pdata v1.33.0
 	go.opentelemetry.io/collector/pdata/testdata v0.127.0
 	go.opentelemetry.io/collector/receiver v1.33.0
@@ -104,8 +106,9 @@ require (
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/collector/config/configcompression v1.33.0 // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.33.0 // indirect
-	go.opentelemetry.io/collector/consumer/consumererror v0.127.0 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.127.0 // indirect
+	go.opentelemetry.io/collector/extension v1.33.0 // indirect
+	go.opentelemetry.io/collector/extension/xextension v0.127.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.33.0 // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.127.0 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.127.0 // indirect

receiver/kafkareceiver/go.sum

@@ -17,6 +17,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -39,8 +40,6 @@ const (
 	attrPartition    = "partition"
 )
 
-var errMemoryLimiterDataRefused = errors.New("data refused due to high memory usage")
-
 type consumeMessageFunc func(ctx context.Context, message *sarama.ConsumerMessage) error
 
 // messageHandler provides a generic interface for handling messages for a pdata type.
@@ -506,7 +505,7 @@ func (c *consumerGroupHandler) handleMessage(
 	c.telemetryBuilder.KafkaReceiverOffsetLag.Record(ctx, claim.HighWaterMarkOffset()-message.Offset-1, metric.WithAttributeSet(attrs))
 
 	if err := c.consumeMessage(ctx, message); err != nil {
-		if errorRequiresBackoff(err) && c.backOff != nil {
+		if c.backOff != nil && !consumererror.IsPermanent(err) {
 			backOffDelay := c.getNextBackoff()
 			if backOffDelay != backoff.Stop {
 				c.logger.Info("Backing off due to error from the next consumer.",
@@ -580,10 +579,6 @@ func newExponentialBackOff(config configretry.BackOffConfig) *backoff.Exponentia
 	return backOff
 }
 
-func errorRequiresBackoff(err error) bool {
-	return err.Error() == errMemoryLimiterDataRefused.Error()
-}
-
 func newContextWithHeaders(ctx context.Context,
 	headers []*sarama.RecordHeader,
 ) context.Context {

receiver/kafkareceiver/kafka_receiver.go

@@ -19,7 +19,9 @@ import (
 	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
@@ -189,15 +191,11 @@ func TestReceiver_ConsumeError(t *testing.T) {
 		shouldRetry bool
 	}{
 		"retryable error": {
-			// FIXME the receiver checks a specific error message
-			// from a different component, which is a bit brittle.
-			// Let's revisit this in the future; we might want to check
-			// for permanent vs. transient errors instead.
-			err:         errMemoryLimiterDataRefused,
+			err:         exporterhelper.ErrQueueIsFull,
 			shouldRetry: true,
 		},
 		"permanent error": {
-			err: errors.New("failed to consume"),
+			err: consumererror.NewPermanent(errors.New("failed to consume")),
 		},
 	} {
 		t.Run(name, func(t *testing.T) {