fix(kafka): ConsumeClaim should not return err when retry recovery fa…

…iled (dapr#1965) * fix(kafka): ConsumeClaim should not return err when retry recovery failed Signed-off-by: zcong1993 <[email protected]> * chore: tweak error log message Signed-off-by: zcong1993 <[email protected]> Signed-off-by: zcong1993 <[email protected]> Co-authored-by: Bernd Verst <[email protected]> Co-authored-by: Dapr Bot <[email protected]> Signed-off-by: Andrew Duss <[email protected]>
mindovermiles262 · Aug 18, 2022 · 1b67380 · 1b67380
1 parent ed9bb52
commit 1b67380
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/internal/component/kafka/consumer.go b/internal/component/kafka/consumer.go
@@ -44,7 +44,7 @@ func (consumer *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
 			}, func() {
 				consumer.k.logger.Infof("Successfully processed Kafka message after it previously failed: %s/%d/%d [key=%s]", message.Topic, message.Partition, message.Offset, asBase64String(message.Key))
 			}); err != nil {
-				return err
+				consumer.k.logger.Errorf("Too many failed attempts at processing Kafka message: %s/%d/%d [key=%s]. Error: %v.", message.Topic, message.Partition, message.Offset, asBase64String(message.Key), err)
 			}
 		} else {
 			err := consumer.doCallback(session, message)