Log processor error in Kafka consumer (#4399)

Signed-off-by: Pavol Loffay <[email protected]> Co-authored-by: Yuri Shkuro <[email protected]>
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()))