Merge pull request #103 from AxonIQ/fix/latency-correctness

Correct latency metrics in AxonIQ Console

Author: CodeDrivenMitch

console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/metrics/AxoniqConsoleProcessorInterceptor.kt

@@ -49,17 +49,18 @@ class AxoniqConsoleProcessorInterceptor(
                 it.reportProcessorName(processorName)
             }
             val segment = unitOfWork.resources()["Processor[$processorName]/SegmentId"] as? Int ?: -1
+            val ingestTimestamp = Instant.now()
             processorMetricsRegistry.registerIngested(
                     processorName,
                     segment,
-                    ChronoUnit.NANOS.between(message.timestamp, Instant.now())
+                    ChronoUnit.NANOS.between(message.timestamp, ingestTimestamp)
             )
             if (unitOfWork !is BatchingUnitOfWork<*> || unitOfWork.isFirstMessage) {
                 unitOfWork.afterCommit {
                     processorMetricsRegistry.registerCommitted(
                             processorName,
                             segment,
-                            ChronoUnit.NANOS.between(message.timestamp, Instant.now())
+                            ChronoUnit.NANOS.between(ingestTimestamp, Instant.now())
                     )
                 }
             }

console-framework-client/src/main/java/io/axoniq/console/framework/eventprocessor/metrics/ProcessorMetricsRegistry.kt

@@ -48,7 +48,7 @@ class ProcessorMetricsRegistry {
         } finally {
             val uow = CurrentUnitOfWork.get()
             if(uow !is BatchingUnitOfWork || uow.isFirstMessage) {
-                uow.afterCommit {
+                uow.onCleanup {
                     getProcessingLatencySegmentMap(processor)
                             .remove(segment)
                 }
@@ -81,7 +81,7 @@ class ProcessorMetricsRegistry {
             .computeIfAbsentWithRetry(processor) { ConcurrentHashMap() }
 
     class ExpiringLatencyValue(
-        private val expiryTime: Long = 30 * 60 * 1000 // Default to 1 hour
+        private val expiryTime: Long = 2 * 60 * 1000 // Default to 2 minutes
     ) {
         private val clock = Clock.systemUTC()
         private val value: AtomicReference<Double> = AtomicReference(-1.0)