Log processor error in Kafka consumer (#4399)

Signed-off-by: Pavol Loffay <p.loffay@gmail.com> Co-authored-by: Yuri Shkuro <yurishkuro@users.noreply.github.com>
jaegertracing · May 26, 2023 · 5274400 · 5274400
1 parent ced77da
commit 5274400
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/cmd/ingester/app/consumer/consumer.go b/cmd/ingester/app/consumer/consumer.go
@@ -146,7 +146,10 @@ func (c *Consumer) handleMessages(pc sc.PartitionConsumer) {
 				defer msgProcessor.Close()
 			}
 
-			msgProcessor.Process(saramaMessageWrapper{msg})
+			err := msgProcessor.Process(saramaMessageWrapper{msg})
+			if err != nil {
+				c.logger.Error("Failed to process a Kafka message", zap.Error(err), zap.Int32("partition", msg.Partition), zap.Int64("offset", msg.Offset))
+			}
 
 		case <-deadlockDetector.closePartitionChannel():
 			c.logger.Info("Closing partition due to inactivity", zap.Int32("partition", pc.Partition()))