Added queue metrics

Robban1980 · Dec 3, 2024 · d06262b · d06262b
1 parent fe40298
commit d06262b
Showing 1 changed file with 5 additions and 0 deletions.
diff --git a/...nt-queue/src/main/java/com/netflix/conductor/kafkaeq/eventqueue/KafkaObservableQueue.java b/...nt-queue/src/main/java/com/netflix/conductor/kafkaeq/eventqueue/KafkaObservableQueue.java
@@ -32,6 +32,7 @@
 import com.netflix.conductor.core.events.queue.Message;
 import com.netflix.conductor.core.events.queue.ObservableQueue;
 import com.netflix.conductor.kafkaeq.config.KafkaEventQueueProperties;
+import com.netflix.conductor.metrics.Monitors;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -164,12 +165,16 @@ public Observable<Message> observe() {
                                                 }
                                             }
 
+                                            Monitors.recordEventQueueMessagesProcessed(
+                                                    QUEUE_TYPE, this.topic, messages.size());
                                             return Observable.from(messages);
                                         } catch (Exception e) {
                                             LOGGER.error(
                                                     "Error while polling Kafka for topic: {}",
                                                     topic,
                                                     e);
+                                            Monitors.recordObservableQMessageReceivedErrors(
+                                                    QUEUE_TYPE);
                                             return Observable.error(e);
                                         }
                                     })