CAMEL-11354: Optimize oldest exchange inflight per route

camel-core/src/main/java/org/apache/camel/Exchange.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel;
 
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 
@@ -584,4 +585,9 @@ public interface Exchange {
      */
     List<Synchronization> handoverCompletions();
 
+    /**
+     * Gets the timestamp when this exchange was created.
+     */
+    Date getCreated();
+
 }

camel-core/src/main/java/org/apache/camel/impl/DefaultExchange.java

@@ -17,6 +17,7 @@
 package org.apache.camel.impl;
 
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedList;
@@ -86,6 +87,15 @@ public String toString() {
         return String.format("Exchange[%s]", exchangeId == null ? "" : exchangeId);
     }
 
+    @Override
+    public Date getCreated() {
+        if (hasProperties()) {
+            return getProperty(Exchange.CREATED_TIMESTAMP, Date.class);
+        } else {
+            return null;
+        }
+    }
+
     public Exchange copy() {
         // to be backwards compatible as today
         return copy(false);

camel-core/src/main/java/org/apache/camel/impl/DefaultInflightRepository.java

@@ -167,7 +167,7 @@ protected void doStop() throws Exception {
 
     private static long getExchangeDuration(Exchange exchange) {
         long duration = 0;
-        Date created = exchange.getProperty(Exchange.CREATED_TIMESTAMP, Date.class);
+        Date created = exchange.getCreated();
         if (created != null) {
             duration = System.currentTimeMillis() - created.getTime();
         }

camel-core/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java

@@ -21,12 +21,12 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.management.AttributeValueExp;
@@ -68,8 +68,7 @@ public class ManagedRoute extends ManagedPerformanceCounter implements TimerList
     protected final String description;
     protected final ModelCamelContext context;
     private final LoadTriplet load = new LoadTriplet();
-    private final ConcurrentSkipListMap<InFlightKey, Long> exchangesInFlightStartTimestamps = new ConcurrentSkipListMap<InFlightKey, Long>();
-    private final ConcurrentHashMap<String, InFlightKey> exchangesInFlightKeys = new ConcurrentHashMap<String, InFlightKey>();
+    private final ConcurrentHashMap<String, Exchange> exchangesInFlight = new ConcurrentHashMap<String, Exchange>();
     private final String jmxDomain;
 
     public ManagedRoute(ModelCamelContext context, Route route) {
@@ -85,8 +84,7 @@ public void init(ManagementStrategy strategy) {
         boolean enabled = context.getManagementStrategy().getManagementAgent().getStatisticsLevel() != ManagementStatisticsLevel.Off;
         setStatisticsEnabled(enabled);
 
-        exchangesInFlightKeys.clear();
-        exchangesInFlightStartTimestamps.clear();
+        exchangesInFlight.clear();
     }
 
     public Route getRoute() {
@@ -410,13 +408,14 @@ public String dumpRouteStatsAsXml(boolean fullStats, boolean includeProcessors)
         String stat = dumpStatsAsXml(fullStats);
         answer.append(" exchangesInflight=\"").append(getInflightExchanges()).append("\"");
         answer.append(" selfProcessingTime=\"").append(routeSelfTime).append("\"");
-        InFlightKey oldestInflightEntry = getOldestInflightEntry();
-        if (oldestInflightEntry == null) {
+        Exchange oldest = getOldestInflightEntry();
+        if (oldest == null) {
             answer.append(" oldestInflightExchangeId=\"\"");
             answer.append(" oldestInflightDuration=\"\"");
         } else {
-            answer.append(" oldestInflightExchangeId=\"").append(oldestInflightEntry.exchangeId).append("\"");
-            answer.append(" oldestInflightDuration=\"").append(System.currentTimeMillis() - oldestInflightEntry.timeStamp).append("\"");
+            long duration = System.currentTimeMillis() - oldest.getCreated().getTime();
+            answer.append(" oldestInflightExchangeId=\"").append(oldest.getExchangeId()).append("\"");
+            answer.append(" oldestInflightDuration=\"").append(duration).append("\"");
         }
         answer.append(" ").append(stat.substring(7, stat.length() - 2)).append(">\n");
 
@@ -466,111 +465,49 @@ public int hashCode() {
         return route.hashCode();
     }
 
-    private InFlightKey getOldestInflightEntry() {
-        Map.Entry<InFlightKey, Long> entry = exchangesInFlightStartTimestamps.firstEntry();
-        if (entry != null) {
-            return entry.getKey();
-        }
-        return null;
+    private Exchange getOldestInflightEntry() {
+        return exchangesInFlight.values().stream().max(Comparator.comparing(Exchange::getCreated)).orElse(null);
     }
 
     public Long getOldestInflightDuration() {
-        InFlightKey oldest = getOldestInflightEntry();
-        if (oldest == null) {
+        Exchange exchange = getOldestInflightEntry();
+        if (exchange == null) {
+            return null;
+        }
+        Date created = exchange.getCreated();
+        if (created != null) {
+            return System.currentTimeMillis() - created.getTime();
+        } else {
             return null;
         }
-        return System.currentTimeMillis() - oldest.timeStamp;
     }
 
     public String getOldestInflightExchangeId() {
-        InFlightKey oldest = getOldestInflightEntry();
-        if (oldest == null) {
+        Exchange exchange = getOldestInflightEntry();
+        if (exchange == null) {
             return null;
         }
-        return oldest.exchangeId;
+        return exchange.getExchangeId();
     }
 
     @Override
     public void processExchange(Exchange exchange) {
-        exchangesInFlightKeys.computeIfAbsent(exchange.getExchangeId(), id -> {
-            InFlightKey key = new InFlightKey(System.currentTimeMillis(), exchange.getExchangeId());
-            exchangesInFlightStartTimestamps.put(key, key.timeStamp);
-            return key;
-        });
+        exchangesInFlight.put(exchange.getExchangeId(), exchange);
         super.processExchange(exchange);
     }
 
     @Override
     public void completedExchange(Exchange exchange, long time) {
-        exchangesInFlightKeys.computeIfPresent(exchange.getExchangeId(), (id, key) -> {
-            exchangesInFlightStartTimestamps.remove(key);
-            return null;
-        });
+        exchangesInFlight.remove(exchange.getExchangeId());
         super.completedExchange(exchange, time);
     }
 
     @Override
     public void failedExchange(Exchange exchange) {
-        exchangesInFlightKeys.computeIfPresent(exchange.getExchangeId(), (id, key) -> {
-            exchangesInFlightStartTimestamps.remove(key);
-            return null;
-        });
+        exchangesInFlight.remove(exchange.getExchangeId());
         super.failedExchange(exchange);
     }
 
-    private static class InFlightKey implements Comparable<InFlightKey> {
-
-        private final Long timeStamp;
-        private final String exchangeId;
-
-        InFlightKey(Long timeStamp, String exchangeId) {
-            this.timeStamp = timeStamp;
-            this.exchangeId = exchangeId;
-        }
-
-        @Override
-        public int compareTo(InFlightKey o) {
-            int compare = Long.compare(timeStamp, o.timeStamp);
-            if (compare == 0) {
-                return exchangeId.compareTo(o.exchangeId);
-            }
-            return compare;
-        }
-
-        @Override
-        public boolean equals(Object o) {
-            if (this == o) {
-                return true;
-            }
-            if (o == null || getClass() != o.getClass()) {
-                return false;
-            }
-
-            InFlightKey that = (InFlightKey) o;
-
-            if (!exchangeId.equals(that.exchangeId)) {
-                return false;
-            }
-            if (!timeStamp.equals(that.timeStamp)) {
-                return false;
-            }
-
-            return true;
-        }
-
-        @Override
-        public int hashCode() {
-            int result = timeStamp.hashCode();
-            result = 31 * result + exchangeId.hashCode();
-            return result;
-        }
-
-        @Override
-        public String toString() {
-            return exchangeId;
-        }
-    }
-
     /**
      * Used for sorting the processor mbeans accordingly to their index.
      */

camel-core/src/main/java/org/apache/camel/util/MessageHelper.java

@@ -560,7 +560,7 @@ public static String doDumpMessageHistoryStacktrace(Exchange exchange, ExchangeF
             label = URISupport.sanitizeUri(exchange.getFromEndpoint().getEndpointUri());
         }
         long elapsed = 0;
-        Date created = exchange.getProperty(Exchange.CREATED_TIMESTAMP, Date.class);
+        Date created = exchange.getCreated();
         if (created != null) {
             elapsed = new StopWatch(created).taken();
         }

camel-core/src/test/java/org/apache/camel/management/EventNotifierExchangeCompletedTest.java

@@ -85,7 +85,7 @@ public void testExchangeCompleted() throws Exception {
         assertEquals("direct://start", event.getExchange().getFromEndpoint().getEndpointUri());
 
         // grab the created timestamp
-        Date created = event.getExchange().getProperty(Exchange.CREATED_TIMESTAMP, Date.class);
+        Date created = event.getExchange().getCreated();
         assertNotNull(created);
 
         // calculate elapsed time