NMS-12987: report currentTimeMillis - flow.lastSwitched as kafka_drif…

…t (watermarks are also based on lastSwitched)
OpenNMS · Jun 4, 2021 · bbc2bbf · bbc2bbf
1 parent ff85cb5
commit bbc2bbf
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/main/src/main/java/org/opennms/nephron/Pipeline.java b/main/src/main/java/org/opennms/nephron/Pipeline.java
@@ -340,6 +340,7 @@ public static class ReadFromKafka extends PTransform<PBegin, PCollection<FlowDoc
         private final String topic;
         private final Map<String, Object> kafkaConsumerConfig;
 
+        // metric name: flink_taskmanager_job_task_operator_flows_from_kafka_drift
         private final Counter flowsFromKafka = Metrics.counter("flows", "from_kafka");
         // a distribution would be more interesting for from_kafka_drift
         // -> Unfortunately histograms are not supported Beam/Flink/Prometheus
@@ -379,7 +380,7 @@ public void processElement(ProcessContext c) {
 
                             // Metrics
                             flowsFromKafka.inc();
-                            flowsFromKafkaDrift.set(System.currentTimeMillis() - flow.getTimestamp());
+                            flowsFromKafkaDrift.set(System.currentTimeMillis() - flow.getLastSwitched().getValue());
                         }
                     }));
         }