From 65b67fbf6ff5ee7f45ce3ec7bbc2f3df966d610b Mon Sep 17 00:00:00 2001
From: Alex Chihaia <chihaiaalex@gmail.com>
Date: Fri, 6 Nov 2026 10:03:25 +0200
Subject: [PATCH] Measure trip update latency from pipeline-ingress timestamp

---
 pom.xml                                                   | 2 +-
 .../transitdata/tripupdate/application/MessageRouter.java | 8 +++-----
 2 files changed, 4 insertions(+), 6 deletions(-)

diff --git a/pom.xml b/pom.xml
index 7eec566..a923ec8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -15,7 +15,7 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.release>25</maven.compiler.release>
-        <common.version>3.1.0</common.version>
+        <common.version>3.2.0</common.version>
         <testcontainers.version>1.21.4</testcontainers.version>
         <spotlessMavenPlugin.version>3.10.3</spotlessMavenPlugin.version>
     </properties>
diff --git a/src/main/java/fi/hsl/transitdata/tripupdate/application/MessageRouter.java b/src/main/java/fi/hsl/transitdata/tripupdate/application/MessageRouter.java
index 5ea682d..c40e367 100644
--- a/src/main/java/fi/hsl/transitdata/tripupdate/application/MessageRouter.java
+++ b/src/main/java/fi/hsl/transitdata/tripupdate/application/MessageRouter.java
@@ -115,7 +115,7 @@ public void handleMessage(Message received) {
 
                             if (tripUpdateIsValid) {
                                 long eventTimeMs = received.getEventTime();
-                                sendTripUpdate(pair, eventTimeMs);
+                                sendTripUpdate(pair, eventTimeMs, received);
                             }
                         } else {
                             log.warn("Failed to process TripUpdate from source schema {}", schema.schema.toString());
@@ -146,7 +146,7 @@ public void handleMessage(Message received) {
     }
 
     private void sendTripUpdate(final AbstractMessageProcessor.TripUpdateWithId tuIdPair,
-            final long pulsarEventTimestamp) {
+            final long pulsarEventTimestamp, final Message received) {
         messageStats.incrementMessagesSent();
 
         final String tripId = tuIdPair.getTripId();
@@ -163,9 +163,7 @@ private void sendTripUpdate(final AbstractMessageProcessor.TripUpdateWithId tuId
                             tripId, tripUpdate.getStopTimeUpdateCount(),
                             tripUpdate.getTrip().getScheduleRelationship()));
 
-            if (pulsarEventTimestamp > 0) {
-                PipelineMetrics.observeLatency(PIPELINE, (System.currentTimeMillis() - pulsarEventTimestamp) / 1000.0);
-            }
+            PipelineMetrics.observeLatencyFromSource(PIPELINE, received);
         });
     }
 }