[receiver/kafka] backoff in case of next consumer error

receiver/kafkareceiver/README.md

@@ -97,7 +97,13 @@ The following settings can be optionally configured:
   - `extract_headers` (default = false): Allows user to attach header fields to resource attributes in otel piepline
   - `headers` (default = []): List of headers they'd like to extract from kafka record. 
   **Note: Matching pattern will be `exact`. Regexes are not supported as of now.** 
-
+- `error_backoff`:
+  - `enabled`: (default = false) Whether to enable backoff when next consumers return errors 
+  - `initial_interval`: The time to wait after the first error before consuming the next message
+  - `max_interval`: The upper bound on backoff interval between consecutive message consumption
+  - `multiplier`: The value multiplied by the backoff interval bounds
+  - `randomization_factor`: A random factor used to calculate next backoff. Randomized interval = RetryInterval * (1 ± RandomizationFactor)
+  - `max_elapsed_time`: The maximum time trying to backoff before giving up. If set to 0, the backoff is never stopped.
 Example:
 
 ```yaml

receiver/kafkareceiver/config.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configretry"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka"
@@ -85,6 +86,9 @@ type Config struct {
 	DefaultFetchSize int32 `mapstructure:"default_fetch_size"`
 	// The maximum bytes per fetch from Kafka (default "0", no limit)
 	MaxFetchSize int32 `mapstructure:"max_fetch_size"`
+
+	// In case of some errors returned by the next consumer, the receiver will wait before consuming the next message
+	ErrorBackOff configretry.BackOffConfig `mapstructure:"error_backoff"`
 }
 
 const (

receiver/kafkareceiver/config_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
 
@@ -65,6 +66,9 @@ func TestLoadConfig(t *testing.T) {
 				MinFetchSize:     1,
 				DefaultFetchSize: 1048576,
 				MaxFetchSize:     0,
+				ErrorBackOff: configretry.BackOffConfig{
+					Enabled: false,
+				},
 			},
 		},
 		{
@@ -101,6 +105,13 @@ func TestLoadConfig(t *testing.T) {
 				MinFetchSize:     1,
 				DefaultFetchSize: 1048576,
 				MaxFetchSize:     0,
+				ErrorBackOff: configretry.BackOffConfig{
+					Enabled:         true,
+					InitialInterval: 1 * time.Second,
+					MaxInterval:     10 * time.Second,
+					MaxElapsedTime:  1 * time.Minute,
+					Multiplier:      1.5,
+				},
 			},
 		},
 	}

receiver/kafkareceiver/kafka_receiver.go

@@ -9,9 +9,12 @@ import (
 	"fmt"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/IBM/sarama"
+	"github.com/cenkalti/backoff/v4"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -35,6 +38,8 @@ const (
 
 var errInvalidInitialOffset = errors.New("invalid initial offset")
 
+var errMemoryLimiterDataRefused = errors.New("data refused due to high memory usage")
+
 // kafkaTracesConsumer uses sarama to consume and handle messages from kafka.
 type kafkaTracesConsumer struct {
 	config            Config
@@ -205,6 +210,7 @@ func (c *kafkaTracesConsumer) Start(_ context.Context, host component.Host) erro
 		messageMarking:    c.messageMarking,
 		headerExtractor:   &nopHeaderExtractor{},
 		telemetryBuilder:  c.telemetryBuilder,
+		backOff:           newExponentialBackOff(c.config.ErrorBackOff),
 	}
 	if c.headerExtraction {
 		consumerGroup.headerExtractor = &headerExtractor{
@@ -313,6 +319,7 @@ func (c *kafkaMetricsConsumer) Start(_ context.Context, host component.Host) err
 		messageMarking:    c.messageMarking,
 		headerExtractor:   &nopHeaderExtractor{},
 		telemetryBuilder:  c.telemetryBuilder,
+		backOff:           newExponentialBackOff(c.config.ErrorBackOff),
 	}
 	if c.headerExtraction {
 		metricsConsumerGroup.headerExtractor = &headerExtractor{
@@ -424,6 +431,7 @@ func (c *kafkaLogsConsumer) Start(_ context.Context, host component.Host) error
 		messageMarking:    c.messageMarking,
 		headerExtractor:   &nopHeaderExtractor{},
 		telemetryBuilder:  c.telemetryBuilder,
+		backOff:           newExponentialBackOff(c.config.ErrorBackOff),
 	}
 	if c.headerExtraction {
 		logsConsumerGroup.headerExtractor = &headerExtractor{
@@ -481,6 +489,7 @@ type tracesConsumerGroupHandler struct {
 	autocommitEnabled bool
 	messageMarking    MessageMarking
 	headerExtractor   HeaderExtractor
+	backOff           *backoff.ExponentialBackOff
 }
 
 type metricsConsumerGroupHandler struct {
@@ -498,6 +507,7 @@ type metricsConsumerGroupHandler struct {
 	autocommitEnabled bool
 	messageMarking    MessageMarking
 	headerExtractor   HeaderExtractor
+	backOff           *backoff.ExponentialBackOff
 }
 
 type logsConsumerGroupHandler struct {
@@ -515,6 +525,7 @@ type logsConsumerGroupHandler struct {
 	autocommitEnabled bool
 	messageMarking    MessageMarking
 	headerExtractor   HeaderExtractor
+	backOff           *backoff.ExponentialBackOff
 }
 
 var (
@@ -582,8 +593,22 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
+				if errorRequiresBackoff(err) && c.backOff != nil {
+					backOffDelay := c.backOff.NextBackOff()
+					if backOffDelay == backoff.Stop {
+						return err
+					}
+					select {
+					case <-session.Context().Done():
+						return nil
+					case <-time.After(backOffDelay):
+					}
+				}
 				return err
 			}
+			if c.backOff != nil {
+				c.backOff.Reset()
+			}
 			if c.messageMarking.After {
 				session.MarkMessage(message, "")
 			}
@@ -600,6 +625,10 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 	}
 }
 
+func errorRequiresBackoff(err error) bool {
+	return errors.Is(err, errMemoryLimiterDataRefused)
+}
+
 func (c *metricsConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
 	c.readyCloser.Do(func() {
 		close(c.ready)
@@ -659,8 +688,22 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
+				if errorRequiresBackoff(err) && c.backOff != nil {
+					backOffDelay := c.backOff.NextBackOff()
+					if backOffDelay == backoff.Stop {
+						return err
+					}
+					select {
+					case <-session.Context().Done():
+						return nil
+					case <-time.After(backOffDelay):
+					}
+				}
 				return err
 			}
+			if c.backOff != nil {
+				c.backOff.Reset()
+			}
 			if c.messageMarking.After {
 				session.MarkMessage(message, "")
 			}
@@ -735,8 +778,22 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
+				if errorRequiresBackoff(err) && c.backOff != nil {
+					backOffDelay := c.backOff.NextBackOff()
+					if backOffDelay == backoff.Stop {
+						return err
+					}
+					select {
+					case <-session.Context().Done():
+						return nil
+					case <-time.After(backOffDelay):
+					}
+				}
 				return err
 			}
+			if c.backOff != nil {
+				c.backOff.Reset()
+			}
 			if c.messageMarking.After {
 				session.MarkMessage(message, "")
 			}
@@ -753,6 +810,20 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 	}
 }
 
+func newExponentialBackOff(config configretry.BackOffConfig) *backoff.ExponentialBackOff {
+	if !config.Enabled {
+		return nil
+	}
+	backOff := backoff.NewExponentialBackOff()
+	backOff.InitialInterval = config.InitialInterval
+	backOff.RandomizationFactor = config.RandomizationFactor
+	backOff.Multiplier = config.Multiplier
+	backOff.MaxInterval = config.MaxInterval
+	backOff.MaxElapsedTime = config.MaxElapsedTime
+	backOff.Reset()
+	return backOff
+}
+
 func toSaramaInitialOffset(initialOffset string) (int64, error) {
 	switch initialOffset {
 	case offsetEarliest: