[receiver/kafkareceiver] Make calls to error backoff thread-safe, add logging (#38941)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
The implementation of
https://pkg.go.dev/github.com/cenkalti/backoff/v4#ExponentialBackOff is
not thread-safe.
When error backoff of kafka receiver is enabled, reading from a topic
with multiple partitions under high load could cause data race error.

This PR protect access to backoff with a mutex to make it thread-safe.

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Start a kafka locally
```
docker run -p 9092:9092 apache/kafka:3.9.0
```

Scale `otlp_logs` topic to 8 partitions
```
./kafka-topics --alter --partitions 8 --topic otlp_logs --bootstrap-server localhost:9092
```

Run the collector locally with the following config
```
# This is a sample collector configuration file. It is also used when running the collector via make run.

receivers:
  kafka:
    encoding: text_utf-8
    error_backoff:
      enabled: true
      initial_interval: 500ms
      max_interval: 10s
      multiplier: 1.5
      randomization_factor: 0
      max_elapsed_time: 1m

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 20
    spike_limit_mib: 0

  batch:

exporters:
  debug:

service:
    logs:
      receivers:
      - kafka
      processors:
      - memory_limiter
      - batch
      exporters:
      - debug
```

Send messages to the kafka topic
```
./kafka-producer-perf-test \
  --topic otlp_logs \
  --throughput 100000 \
  --num-records 1000000 \
  --record-size 5024 \
  --producer-props bootstrap.servers=localhost:9092
```

From the log messages we can better observe the backoff behavior

Author: yiquanzhou

.chloggen/improve-kafka-error-backoff.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: make calls to error backoff thread-safe and add logging
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38941]
+
+# (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/kafka_receiver.go

@@ -460,6 +460,7 @@ type tracesConsumerGroupHandler struct {
 	messageMarking    MessageMarking
 	headerExtractor   HeaderExtractor
 	backOff           *backoff.ExponentialBackOff
+	backOffMutex      sync.Mutex
 }
 
 type metricsConsumerGroupHandler struct {
@@ -478,6 +479,7 @@ type metricsConsumerGroupHandler struct {
 	messageMarking    MessageMarking
 	headerExtractor   HeaderExtractor
 	backOff           *backoff.ExponentialBackOff
+	backOffMutex      sync.Mutex
 }
 
 type logsConsumerGroupHandler struct {
@@ -496,6 +498,7 @@ type logsConsumerGroupHandler struct {
 	messageMarking    MessageMarking
 	headerExtractor   HeaderExtractor
 	backOff           *backoff.ExponentialBackOff
+	backOffMutex      sync.Mutex
 }
 
 var (
@@ -561,8 +564,13 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 			c.obsrecv.EndTracesOp(ctx, c.unmarshaler.Encoding(), spanCount, err)
 			if err != nil {
 				if errorRequiresBackoff(err) && c.backOff != nil {
-					backOffDelay := c.backOff.NextBackOff()
+					backOffDelay := c.getNextBackoff()
 					if backOffDelay != backoff.Stop {
+						c.logger.Info("Backing off due to error from the next consumer.",
+							zap.Error(err),
+							zap.Duration("delay", backOffDelay),
+							zap.String("topic", message.Topic),
+							zap.Int32("partition", claim.Partition()))
 						select {
 						case <-session.Context().Done():
 							return nil
@@ -574,14 +582,16 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 							return err
 						}
 					}
+					c.logger.Info("Stop error backoff because the configured max_elapsed_time is reached",
+						zap.Duration("max_elapsed_time", c.backOff.MaxElapsedTime))
 				}
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
 				return err
 			}
 			if c.backOff != nil {
-				c.backOff.Reset()
+				c.resetBackoff()
 			}
 			if c.messageMarking.After {
 				session.MarkMessage(message, "")
@@ -599,6 +609,18 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 	}
 }
 
+func (c *tracesConsumerGroupHandler) getNextBackoff() time.Duration {
+	c.backOffMutex.Lock()
+	defer c.backOffMutex.Unlock()
+	return c.backOff.NextBackOff()
+}
+
+func (c *tracesConsumerGroupHandler) resetBackoff() {
+	c.backOffMutex.Lock()
+	defer c.backOffMutex.Unlock()
+	c.backOff.Reset()
+}
+
 func (c *metricsConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
 	c.readyCloser.Do(func() {
 		close(c.ready)
@@ -656,8 +678,13 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 			c.obsrecv.EndMetricsOp(ctx, c.unmarshaler.Encoding(), dataPointCount, err)
 			if err != nil {
 				if errorRequiresBackoff(err) && c.backOff != nil {
-					backOffDelay := c.backOff.NextBackOff()
+					backOffDelay := c.getNextBackoff()
 					if backOffDelay != backoff.Stop {
+						c.logger.Info("Backing off due to error from the next consumer.",
+							zap.Error(err),
+							zap.Duration("delay", backOffDelay),
+							zap.String("topic", message.Topic),
+							zap.Int32("partition", claim.Partition()))
 						select {
 						case <-session.Context().Done():
 							return nil
@@ -669,14 +696,16 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 							return err
 						}
 					}
+					c.logger.Info("Stop error backoff because the configured max_elapsed_time is reached",
+						zap.Duration("max_elapsed_time", c.backOff.MaxElapsedTime))
 				}
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
 				return err
 			}
 			if c.backOff != nil {
-				c.backOff.Reset()
+				c.resetBackoff()
 			}
 			if c.messageMarking.After {
 				session.MarkMessage(message, "")
@@ -694,6 +723,18 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 	}
 }
 
+func (c *metricsConsumerGroupHandler) getNextBackoff() time.Duration {
+	c.backOffMutex.Lock()
+	defer c.backOffMutex.Unlock()
+	return c.backOff.NextBackOff()
+}
+
+func (c *metricsConsumerGroupHandler) resetBackoff() {
+	c.backOffMutex.Lock()
+	defer c.backOffMutex.Unlock()
+	c.backOff.Reset()
+}
+
 func (c *logsConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
 	c.readyCloser.Do(func() {
 		close(c.ready)
@@ -750,8 +791,13 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 			c.obsrecv.EndLogsOp(ctx, c.unmarshaler.Encoding(), logRecordCount, err)
 			if err != nil {
 				if errorRequiresBackoff(err) && c.backOff != nil {
-					backOffDelay := c.backOff.NextBackOff()
+					backOffDelay := c.getNextBackoff()
 					if backOffDelay != backoff.Stop {
+						c.logger.Info("Backing off due to error from the next consumer.",
+							zap.Error(err),
+							zap.Duration("delay", backOffDelay),
+							zap.String("topic", message.Topic),
+							zap.Int32("partition", claim.Partition()))
 						select {
 						case <-session.Context().Done():
 							return nil
@@ -763,14 +809,16 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 							return err
 						}
 					}
+					c.logger.Info("Stop error backoff because the configured max_elapsed_time is reached",
+						zap.Duration("max_elapsed_time", c.backOff.MaxElapsedTime))
 				}
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
 				return err
 			}
 			if c.backOff != nil {
-				c.backOff.Reset()
+				c.resetBackoff()
 			}
 			if c.messageMarking.After {
 				session.MarkMessage(message, "")
@@ -788,6 +836,18 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 	}
 }
 
+func (c *logsConsumerGroupHandler) getNextBackoff() time.Duration {
+	c.backOffMutex.Lock()
+	defer c.backOffMutex.Unlock()
+	return c.backOff.NextBackOff()
+}
+
+func (c *logsConsumerGroupHandler) resetBackoff() {
+	c.backOffMutex.Lock()
+	defer c.backOffMutex.Unlock()
+	c.backOff.Reset()
+}
+
 func newExponentialBackOff(config configretry.BackOffConfig) *backoff.ExponentialBackOff {
 	if !config.Enabled {
 		return nil