retry error again

Author: yiquanzhou

receiver/kafkareceiver/config.go

@@ -86,7 +86,7 @@ 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
+	// In case of some errors returned by the next consumer, the receiver will wait and retry the failed message
 	ErrorBackOff configretry.BackOffConfig `mapstructure:"error_backoff"`
 }
 

receiver/kafkareceiver/kafka_receiver.go

@@ -552,9 +552,6 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 	if !c.autocommitEnabled {
 		defer session.Commit()
 	}
-	if c.backOff != nil {
-		c.backOff.Reset()
-	}
 	for {
 		select {
 		case message, ok := <-claim.Messages():
@@ -600,6 +597,11 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 						case <-session.Context().Done():
 							return nil
 						case <-time.After(backOffDelay):
+							if !c.messageMarking.After {
+								// Unmark the message so it can be retried
+								session.ResetOffset(claim.Topic(), claim.Partition(), message.Offset, "")
+							}
+							return err
 						}
 					}
 				}
@@ -627,10 +629,6 @@ 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)
@@ -649,9 +647,6 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 	if !c.autocommitEnabled {
 		defer session.Commit()
 	}
-	if c.backOff != nil {
-		c.backOff.Reset()
-	}
 	for {
 		select {
 		case message, ok := <-claim.Messages():
@@ -697,6 +692,11 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 						case <-session.Context().Done():
 							return nil
 						case <-time.After(backOffDelay):
+							if !c.messageMarking.After {
+								// Unmark the message so it can be retried
+								session.ResetOffset(claim.Topic(), claim.Partition(), message.Offset, "")
+							}
+							return err
 						}
 					}
 				}
@@ -786,6 +786,11 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 						case <-session.Context().Done():
 							return nil
 						case <-time.After(backOffDelay):
+							if !c.messageMarking.After {
+								// Unmark the message so it can be retried
+								session.ResetOffset(claim.Topic(), claim.Partition(), message.Offset, "")
+							}
+							return err
 						}
 					}
 				}
@@ -866,3 +871,7 @@ func encodingToComponentID(encoding string) (*component.ID, error) {
 	id := component.NewID(componentType)
 	return &id, nil
 }
+
+func errorRequiresBackoff(err error) bool {
+	return err.Error() == errMemoryLimiterDataRefused.Error()
+}