retry error again

Author: yiquanzhou

receiver/kafkareceiver/kafka_receiver.go

@@ -600,6 +600,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
 						}
 					}
 				}
@@ -628,7 +633,7 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 }
 
 func errorRequiresBackoff(err error) bool {
-	return errors.Is(err, errMemoryLimiterDataRefused)
+	return err.Error() == errMemoryLimiterDataRefused.Error()
 }
 
 func (c *metricsConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
@@ -649,9 +654,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 +699,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 +793,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
 						}
 					}
 				}