Report error when Kafka connection is broken (#615)

netobserv · Feb 28, 2024 · a140b7f · a140b7f
1 parent 817e677
commit a140b7f
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/pkg/pipeline/ingest/ingest_kafka.go b/pkg/pipeline/ingest/ingest_kafka.go
@@ -87,6 +87,7 @@ func (k *ingestKafka) kafkaListener() {
 			kafkaMessage, err := k.kafkaReader.ReadMessage(context.Background())
 			if err != nil {
 				klog.Errorln(err)
+				k.metrics.error("Cannot read message")
 				continue
 			}
 			if k.canLogMessages && logrus.IsLevelEnabled(logrus.TraceLevel) {