This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 7315c70a249b CAMEL-26393: camel-management - JMX performance counters:
reduce contention on the per-exchange path (#28760)
7315c70a249b is described below
commit 7315c70a249b3de311ee6a0a1771a265a4ee3298
Author: Adriano Machado <[email protected]>
AuthorDate: Sat Oct 31 05:29:16 2026 -0400
CAMEL-26393: camel-management - JMX performance counters: reduce contention
on the per-exchange path (#28760)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../management/mbean/ManagedCamelContext.java | 2 +-
.../mbean/ManagedPerformanceCounter.java | 27 ++-
.../camel/management/mbean/ManagedRoute.java | 2 +-
.../camel/management/mbean/StatisticCounter.java | 16 +-
.../camel/management/mbean/StatisticDelta.java | 10 +-
.../camel/management/mbean/StatisticValue.java | 5 +-
.../camel/management/ManagedStatisticsTest.java | 4 +-
.../mbean/ManagedPerformanceCounterTest.java | 202 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 +
9 files changed, 247 insertions(+), 28 deletions(-)
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
index 342212324621..c4ca9819ae53 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedCamelContext.java
@@ -408,7 +408,7 @@ public class ManagedCamelContext extends
ManagedPerformanceCounter implements Ma
@Override
public long getRemoteExchangesInflight() {
- return remoteExchangesInflight.getValue();
+ return Math.max(0, remoteExchangesInflight.getValue());
}
@Override
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
index e0bf112a7c0f..1e26083c5930 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
@@ -57,7 +57,6 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
private Statistic totalProcessingTime;
private Statistic lastProcessingTime;
private Statistic deltaProcessingTime;
- private Statistic meanProcessingTime;
private Statistic firstExchangeCompletedTimestamp;
private String firstExchangeCompletedExchangeId;
private Statistic firstExchangeFailureTimestamp;
@@ -92,7 +91,6 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
this.totalProcessingTime = new StatisticCounter();
this.lastProcessingTime = new StatisticValue();
this.deltaProcessingTime = new StatisticDelta();
- this.meanProcessingTime = new StatisticValue();
this.firstExchangeCompletedTimestamp = new StatisticValue();
this.firstExchangeFailureTimestamp = new StatisticValue();
@@ -122,7 +120,6 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
totalProcessingTime.reset();
lastProcessingTime.reset();
deltaProcessingTime.reset();
- meanProcessingTime.reset();
firstExchangeCompletedTimestamp.reset();
firstExchangeCompletedExchangeId = null;
firstExchangeFailureTimestamp.reset();
@@ -152,7 +149,8 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
@Override
public long getExchangesInflight() {
- return exchangesInflight.getValue();
+ // the sum of the counter can see a decrement before its increment
+ return Math.max(0, exchangesInflight.getValue());
}
@Override
@@ -177,7 +175,9 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
@Override
public long getMeanProcessingTime() {
- return meanProcessingTime.getValue();
+ // computed here as summing the counters is too costly to do for every
exchange
+ long completed = exchangesCompleted.getValue();
+ return completed > 0 ? totalProcessingTime.getValue() / completed : -1;
}
@Override
@@ -374,19 +374,14 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
firstExchangeCompletedTimestamp.updateValue(now);
}
- lastExchangeCompletedTimestamp.updateValue(now);
if (firstExchangeCompletedExchangeId == null) {
firstExchangeCompletedExchangeId = exchange.getExchangeId();
}
- lastExchangeCompletedExchangeId = exchange.getExchangeId();
-
- // update mean
- long mean = 0;
- long completed = exchangesCompleted.getValue();
- if (completed > 0) {
- mean = totalProcessingTime.getValue() / completed;
+ // at most once per millisecond, as getting the id generates it for an
exchange that has none yet
+ if (lastExchangeCompletedTimestamp.getValue() != now) {
+ lastExchangeCompletedExchangeId = exchange.getExchangeId();
+ lastExchangeCompletedTimestamp.updateValue(now);
}
- meanProcessingTime.updateValue(mean);
}
@Override
@@ -445,7 +440,7 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
sb.append(String.format(" totalProcessingTime=\"%s\"",
totalProcessingTime.getValue()));
sb.append(String.format(" lastProcessingTime=\"%s\"",
lastProcessingTime.getValue()));
sb.append(String.format(" deltaProcessingTime=\"%s\"",
deltaProcessingTime.getValue()));
- sb.append(String.format(" meanProcessingTime=\"%s\"",
meanProcessingTime.getValue()));
+ sb.append(String.format(" meanProcessingTime=\"%s\"",
getMeanProcessingTime()));
sb.append(String.format(" exchangesThroughput=\"%s\"",
getThroughput()));
if (percentileWindow != null) {
sb.append(String.format(" p50ProcessingTime=\"%s\"",
getProcessingTimeP50()));
@@ -498,7 +493,7 @@ public abstract class ManagedPerformanceCounter extends
ManagedCounter
jo.put("totalProcessingTime", totalProcessingTime.getValue());
jo.put("lastProcessingTime", lastProcessingTime.getValue());
jo.put("deltaProcessingTime", deltaProcessingTime.getValue());
- jo.put("meanProcessingTime", meanProcessingTime.getValue());
+ jo.put("meanProcessingTime", getMeanProcessingTime());
jo.put("exchangesThroughput", getThroughput());
if (percentileWindow != null) {
jo.put("p50ProcessingTime", getProcessingTimeP50());
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
index cff094fc7858..be08000140f2 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
@@ -895,7 +895,7 @@ public class ManagedRoute extends ManagedPerformanceCounter
implements ManagedRo
@Override
public long getRemoteExchangesInflight() {
- return remoteConsumer ? getExchangesInflight() :
remoteExchangesInflight.getValue();
+ return remoteConsumer ? getExchangesInflight() : Math.max(0,
remoteExchangesInflight.getValue());
}
@Override
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticCounter.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticCounter.java
index cf0b1a6914dd..c96ed77fd542 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticCounter.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticCounter.java
@@ -16,30 +16,34 @@
*/
package org.apache.camel.management.mbean;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+/**
+ * A counter updated for every exchange by many threads, so it is striped:
updates scale, and the value is summed when
+ * it is read. A value read during concurrent updates can miss an update that
is in progress.
+ */
public class StatisticCounter extends Statistic {
- private final AtomicLong value = new AtomicLong();
+ private final LongAdder value = new LongAdder();
@Override
public void updateValue(long newValue) {
- value.getAndAdd(newValue);
+ value.add(newValue);
}
@Override
public long getValue() {
- return value.get();
+ return value.sum();
}
@Override
public String toString() {
- return Long.toString(value.get());
+ return Long.toString(value.sum());
}
@Override
public void reset() {
- value.set(0);
+ value.reset();
}
@Override
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticDelta.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticDelta.java
index edbd18a18e5a..bb79656116eb 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticDelta.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticDelta.java
@@ -25,8 +25,14 @@ public class StatisticDelta extends Statistic {
@Override
public void updateValue(long newValue) {
- lastValue.set(value.longValue());
- value.set(newValue);
+ // a read leaves the cache line shared between threads, which a write
for an unchanged value does not
+ long current = value.get();
+ if (lastValue.get() != current) {
+ lastValue.set(current);
+ }
+ if (current != newValue) {
+ value.set(newValue);
+ }
}
@Override
diff --git
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticValue.java
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticValue.java
index 1d4686b6f80f..4b42848458dc 100644
---
a/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticValue.java
+++
b/core/camel-management/src/main/java/org/apache/camel/management/mbean/StatisticValue.java
@@ -24,7 +24,10 @@ public class StatisticValue extends Statistic {
@Override
public void updateValue(long newValue) {
- value.set(newValue);
+ // a read leaves the cache line shared between threads, which a write
for an unchanged value does not
+ if (value.get() != newValue) {
+ value.set(newValue);
+ }
}
@Override
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedStatisticsTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedStatisticsTest.java
index 968e5454c569..62a21a726657 100644
---
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedStatisticsTest.java
+++
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedStatisticsTest.java
@@ -30,6 +30,7 @@ import org.junit.jupiter.api.condition.OS;
import static
org.apache.camel.management.DefaultManagementObjectNameStrategy.TYPE_PROCESSOR;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
@DisabledOnOs(OS.AIX)
public class ManagedStatisticsTest extends ManagementTestSupport {
@@ -71,7 +72,8 @@ public class ManagedStatisticsTest extends
ManagementTestSupport {
assertNull(firstFail);
String last = (String) mbeanServer.getAttribute(on,
"LastExchangeCompletedExchangeId");
- assertEquals(result.getReceivedExchanges().get(4).getExchangeId(),
last);
+ // updated at most once per millisecond, and the exchanges can be sent
within the same millisecond
+ assertTrue(result.getReceivedExchanges().stream().anyMatch(e ->
e.getExchangeId().equals(last)));
String lastFail = (String) mbeanServer.getAttribute(on,
"LastExchangeFailureExchangeId");
assertNull(lastFail);
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/mbean/ManagedPerformanceCounterTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/mbean/ManagedPerformanceCounterTest.java
new file mode 100644
index 000000000000..956fca9195f2
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/mbean/ManagedPerformanceCounterTest.java
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.management.mbean;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ManagedPerformanceCounterTest {
+
+ private static final int THREADS = 8;
+ private static final int UPDATES = 20000;
+
+ private CamelContext context;
+ private ManagedPerformanceCounter counter;
+
+ @BeforeEach
+ public void setUp() {
+ context = new DefaultCamelContext();
+ counter = new ManagedPerformanceCounter() {
+ };
+ counter.init(context.getManagementStrategy());
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ context.close();
+ }
+
+ @Test
+ public void testMeanIsUnsetUntilAnExchangeCompletes() {
+ // monitoring tools rely on -1 meaning no data, which is not the same
as a mean of 0 ms
+ assertEquals(-1, counter.getMeanProcessingTime());
+
+ counter.failedExchange(new DefaultExchange(context));
+ assertEquals(-1, counter.getMeanProcessingTime());
+
+ counter.completedExchange(new DefaultExchange(context), 10);
+ counter.completedExchange(new DefaultExchange(context), 20);
+ assertEquals(15, counter.getMeanProcessingTime());
+
+ counter.reset();
+ assertEquals(-1, counter.getMeanProcessingTime());
+ }
+
+ @Test
+ public void testDeltaIsZeroForEqualProcessingTimes() {
+ // repeated values must not leave the delta of the previous change
behind
+ counter.completedExchange(new DefaultExchange(context), 5);
+ assertEquals(5, counter.getDeltaProcessingTime());
+ counter.completedExchange(new DefaultExchange(context), 8);
+ assertEquals(3, counter.getDeltaProcessingTime());
+ counter.completedExchange(new DefaultExchange(context), 8);
+ assertEquals(0, counter.getDeltaProcessingTime());
+ counter.completedExchange(new DefaultExchange(context), 2);
+ assertEquals(-6, counter.getDeltaProcessingTime());
+ }
+
+ @Test
+ public void testInflightSurvivesReset() {
+ // exchanges that are inflight during a reset complete afterwards, and
must still be counted down
+ counter.processExchange(new DefaultExchange(context), "processor");
+ counter.reset();
+ assertEquals(1, counter.getExchangesInflight());
+
+ counter.completedExchange(new DefaultExchange(context), 1);
+ assertEquals(0, counter.getExchangesInflight());
+ }
+
+ @Test
+ public void testInflightIsNeverNegative() {
+ // the sum read over the striped cells of the counter can see a
decrement before its increment
+ counter.completedExchange(new DefaultExchange(context), 1);
+ assertEquals(0, counter.getExchangesInflight());
+ }
+
+ @Test
+ public void testLastExchangeIdDoesNotGenerateAnIdPerExchange() {
+ // exchange ids are lazy and a random id is costly, so the counter
must not ask each exchange for its id
+ AtomicLong generated = new AtomicLong();
+ context.setUuidGenerator(() -> "ID-" + generated.incrementAndGet());
+
+ int exchanges = 50000;
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < exchanges; i++) {
+ counter.completedExchange(new DefaultExchange(context), 0);
+ }
+ long elapsed = System.currentTimeMillis() - start;
+
+ assertEquals(exchanges, counter.getExchangesCompleted());
+ // at most one id for every millisecond the loop was seen in
+ long max = elapsed + 1;
+ assertTrue(generated.get() <= max,
+ "Generated " + generated.get() + " ids in " + elapsed + " ms,
expected at most " + max);
+ }
+
+ @Test
+ public void testLastExchangeIdIsUpdatedInALaterMillisecond() {
+ Exchange first = new DefaultExchange(context);
+ counter.completedExchange(first, 1);
+ assertEquals(first.getExchangeId(),
counter.getFirstExchangeCompletedExchangeId());
+ assertEquals(first.getExchangeId(),
counter.getLastExchangeCompletedExchangeId());
+
+ long time = counter.getLastExchangeCompletedTimestamp().getTime();
+ while (System.currentTimeMillis() <= time) {
+ Thread.onSpinWait();
+ }
+
+ Exchange second = new DefaultExchange(context);
+ counter.completedExchange(second, 1);
+ assertEquals(first.getExchangeId(),
counter.getFirstExchangeCompletedExchangeId());
+ assertEquals(second.getExchangeId(),
counter.getLastExchangeCompletedExchangeId());
+ }
+
+ @Test
+ public void testResetClearsLastExchangeId() {
+ counter.completedExchange(new DefaultExchange(context), 1);
+ counter.reset();
+ assertNull(counter.getLastExchangeCompletedExchangeId());
+
+ // also within the same millisecond as the exchange before the reset
+ Exchange exchange = new DefaultExchange(context);
+ counter.completedExchange(exchange, 1);
+ assertEquals(exchange.getExchangeId(),
counter.getLastExchangeCompletedExchangeId());
+ }
+
+ @Test
+ public void testLastFailedExchangeIdIsExact() {
+ // failures are rare and worth finding, so they are not throttled as
completed exchanges are
+ counter.failedExchange(new DefaultExchange(context));
+ Exchange last = new DefaultExchange(context);
+ counter.failedExchange(last);
+ assertEquals(last.getExchangeId(),
counter.getLastExchangeFailureExchangeId());
+ }
+
+ @Test
+ public void testConcurrentExchangesAreAllCounted() throws Exception {
+ ExecutorService executor = Executors.newFixedThreadPool(THREADS);
+ try {
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<?>> futures = new ArrayList<>();
+ for (int t = 0; t < THREADS; t++) {
+ final long time = t + 1L;
+ futures.add(executor.submit(() -> {
+ start.await();
+ Exchange exchange = new DefaultExchange(context);
+ for (int i = 0; i < UPDATES; i++) {
+ counter.processExchange(exchange, "processor");
+ counter.completedExchange(exchange, time);
+ }
+ return null;
+ }));
+ }
+ start.countDown();
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+
+ assertEquals(160000, counter.getExchangesTotal());
+ assertEquals(160000, counter.getExchangesCompleted());
+ assertEquals(0, counter.getExchangesInflight());
+ // 20000 * (1 + 2 + ... + 8)
+ assertEquals(720000, counter.getTotalProcessingTime());
+ assertEquals(4, counter.getMeanProcessingTime());
+ assertEquals(1, counter.getMinProcessingTime());
+ assertEquals(8, counter.getMaxProcessingTime());
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index d298864d2e54..5e1a0e08c1cc 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -100,6 +100,13 @@ in its model, so clearing them on the route saved little
memory at the cost of m
Applications that called `Route.clearRouteModel()` directly can remove the
call. Implementations of the `Route` interface
that override `clearRouteModel()` can remove the override.
+=== camel-management - LastExchangeCompletedExchangeId is updated at most once
per millisecond
+
+The JMX attribute `LastExchangeCompletedExchangeId`, of the `CamelContext`,
route and processor MBeans, is now updated
+at most once per millisecond, so that the JMX statistics no longer generate an
exchange id for every exchange. When
+several exchanges complete within the same millisecond, the attribute has the
id of the first one, not of the last one.
+Failed exchanges are not affected: `LastExchangeFailureExchangeId` is still
updated for every failed exchange.
+
=== camel-core - RedeliveryErrorHandler.defensiveCopyExchangeIfNeeded is
deprecated
When redelivery is enabled, the error handler (`DefaultErrorHandler`,
`DeadLetterChannel` and the transaction error
