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 7cce4ff2f34a CAMEL-26781: camel-core - the runtime endpoint registry
counts the reply of an InOut consumer once per message (#29478)
7cce4ff2f34a is described below
commit 7cce4ff2f34ac17410bc65cc95083aecafe37242
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 9 22:47:02 2026 +0100
CAMEL-26781: camel-core - the runtime endpoint registry counts the reply of
an InOut consumer once per message (#29478)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../resilience4j/ResilienceInputHitsTest.java | 13 ++-
.../engine/DefaultRuntimeEndpointRegistry.java | 99 +++++++++++---------
.../management/ManagedEndpointReplyHitsTest.java | 100 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 6 +-
4 files changed, 176 insertions(+), 42 deletions(-)
diff --git
a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceInputHitsTest.java
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceInputHitsTest.java
index 2f47bd2129ac..823851a3dd3d 100644
---
a/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceInputHitsTest.java
+++
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceInputHitsTest.java
@@ -27,7 +27,8 @@ import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
- * The circuit breaker runs a copy of the exchange with its own unit of work:
the copy is no second input of the route.
+ * The circuit breaker runs a copy of the exchange with its own unit of work:
the copy is no second input of the route,
+ * and no second reply of an InOut consumer.
*/
public class ResilienceInputHitsTest extends CamelTestSupport {
@@ -55,6 +56,16 @@ public class ResilienceInputHitsTest extends
CamelTestSupport {
assertEquals(3, hits("supplier", "direct://supplier", "in"));
}
+ @Test
+ public void testOneReplyPerRequest() {
+ for (int i = 0; i < 3; i++) {
+ assertEquals("in stock", template.requestBody("seda:check",
"CAMEL-MUG"));
+ }
+
+ assertEquals(3, hits("check", "seda://check", "in"));
+ assertEquals(3, hits("check", "seda://check", "out"));
+ }
+
private long hits(String routeId, String uri, String direction) {
return
context.getRuntimeEndpointRegistry().getEndpointStatistics().stream()
.filter(s -> routeId.equals(s.getRouteId()) &&
uri.equals(s.getUri()) && direction.equals(s.getDirection()))
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRuntimeEndpointRegistry.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRuntimeEndpointRegistry.java
index 83e840dc5c91..1d4c6af21ae4 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRuntimeEndpointRegistry.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRuntimeEndpointRegistry.java
@@ -66,7 +66,7 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
private EndpointUtilizationStatistics outputUtilization;
private EndpointSizeStatistics inputSizeStats;
private EndpointSizeStatistics outputSizeStats;
- private final RoutePolicySupport inputHits = new InputHits();
+ private final RoutePolicySupport consumerHits = new ConsumerHits();
@Override
public boolean isEnabled() {
@@ -215,12 +215,13 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
}
if (extended) {
// the input hits of a route are counted when an exchange enters
it through its consumer, as the route
- // counts its exchanges: the copies EIPs make inside a route
(circuit breaker, split, multicast, recipient
- // list) have their own unit of work and ExchangeCreatedEvent but
are no input, and an exchange that comes
- // from another route over direct is one without an
ExchangeCreatedEvent.
- // One shared instance for all routes: InputHits keeps no
per-route state, the counts live in the
- // thread-safe inputUtilization
- getCamelContext().addRoutePolicyFactory((context, routeId, route)
-> inputHits);
+ // counts its exchanges, and the replies of an InOut consumer when
the exchange is done there: the copies
+ // EIPs make inside a route (circuit breaker, split, multicast,
recipient list) have their own unit of work,
+ // created and completed events, but are no input and no reply,
and an exchange that comes from another
+ // route over direct is one without an ExchangeCreatedEvent.
+ // One shared instance for all routes: ConsumerHits keeps no
per-route state, the counts live in the
+ // thread-safe inputUtilization and outputUtilization
+ getCamelContext().addRoutePolicyFactory((context, routeId, route)
-> consumerHits);
LOG.debug(
"Runtime endpoint registry is in extended mode gathering
usage statistics of all incoming and outgoing endpoints (cache limit: {})",
limit);
@@ -353,36 +354,14 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
exchange.removeProperty("CamelMessageHeadersSizeSending");
}
} else if (event instanceof ExchangeCompletedEvent || event instanceof
ExchangeFailedEvent) {
- // InOut consumers send a reply back when the exchange completes;
- // record this as an "out" hit on the consumer's fromEndpoint
- CamelEvent.ExchangeEvent ee = (CamelEvent.ExchangeEvent) event;
- Exchange exchange = ee.getExchange();
- if (exchange.getPattern() != null &&
exchange.getPattern().isOutCapable()) {
- Endpoint endpoint = exchange.getFromEndpoint();
- if (endpoint != null) {
- String routeId = exchange.getFromRouteId();
- String uri = endpoint.getEndpointUri();
- Map<String, String> uris = outputs.get(routeId);
- if (uris != null) {
- uris.putIfAbsent(uri, uri);
- }
- if (extended) {
- String key = asUtilizationKey(routeId, uri);
- if (key != null) {
- outputUtilization.onHit(key);
- if (messageSizeEnabled) {
- Message message = exchange.getMessage();
- MessageSizeStrategy strategy =
getCamelContext().getMessageSizeStrategy();
- long bodySize =
strategy.computeBodySize(message);
- long headersSize =
strategy.computeHeadersSize(message);
- outputSizeStats.onHit(key, bodySize,
headersSize);
- // reset stream cache so the body is
re-readable
- if (message.getBody() instanceof StreamCache
sc) {
- sc.reset();
- }
- }
- }
- }
+ // an InOut consumer sends a reply back when the exchange
completes: its endpoint is an output of the route
+ // (the hits are counted by the route policy, as the copies EIPs
make complete as well)
+ Exchange exchange = ((CamelEvent.ExchangeEvent)
event).getExchange();
+ Endpoint endpoint = exchange.getFromEndpoint();
+ if (endpoint != null && exchange.getPattern() != null &&
exchange.getPattern().isOutCapable()) {
+ Map<String, String> uris =
outputs.get(exchange.getFromRouteId());
+ if (uris != null) {
+ uris.putIfAbsent(endpoint.getEndpointUri(),
endpoint.getEndpointUri());
}
}
}
@@ -439,8 +418,41 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
}
}
- /** The route policy of every route in extended mode: an exchange that
begins in a route is an input hit. */
- private final class InputHits extends RoutePolicySupport {
+ /**
+ * An InOut consumer sends a reply when the exchange it created is done in
its route: an out hit of the consumer's
+ * endpoint. Not for a route the exchange came to over direct (its
consumer sends no reply of its own), nor for the
+ * copies EIPs make inside a route, which are done in their own unit of
work but never in the route.
+ */
+ private void onRouteReply(Route route, Exchange exchange) {
+ Endpoint endpoint = exchange.getFromEndpoint();
+ String routeId = route.getRouteId();
+ if (endpoint == null || exchange.getPattern() == null ||
!exchange.getPattern().isOutCapable()
+ || !routeId.equals(exchange.getFromRouteId())) {
+ return;
+ }
+ String key = asUtilizationKey(routeId, endpoint.getEndpointUri());
+ if (key == null || outputUtilization == null) {
+ return;
+ }
+ outputUtilization.onHit(key);
+ if (messageSizeEnabled) {
+ Message message = exchange.getMessage();
+ MessageSizeStrategy strategy =
getCamelContext().getMessageSizeStrategy();
+ long bodySize = strategy.computeBodySize(message);
+ long headersSize = strategy.computeHeadersSize(message);
+ outputSizeStats.onHit(key, bodySize, headersSize);
+ // reset stream cache so the body is re-readable
+ if (message.getBody() instanceof StreamCache sc) {
+ sc.reset();
+ }
+ }
+ }
+
+ /**
+ * The route policy of every route in extended mode: an exchange that
begins in a route is an input hit, and the
+ * reply of an InOut consumer when it is done there an out hit.
+ */
+ private final class ConsumerHits extends RoutePolicySupport {
@Override
public void onExchangeBegin(Route route, Exchange exchange) {
@@ -449,9 +461,16 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
}
}
+ @Override
+ public void onExchangeDone(Route route, Exchange exchange) {
+ if (extended && enabled) {
+ onRouteReply(route, exchange);
+ }
+ }
+
@Override
public String toString() {
- return "RuntimeEndpointRegistryInputHits";
+ return "RuntimeEndpointRegistryConsumerHits";
}
}
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedEndpointReplyHitsTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedEndpointReplyHitsTest.java
new file mode 100644
index 000000000000..b1f2369fc33f
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedEndpointReplyHitsTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ManagementStatisticsLevel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.RuntimeEndpointRegistry;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * An InOut consumer sends a reply per message it received, as an out hit of
its endpoint: the sub-exchanges of a split,
+ * multicast or recipient list are no reply, and a route called over direct
sends none.
+ */
+@DisabledOnOs(OS.AIX)
+public class ManagedEndpointReplyHitsTest extends ManagementTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+
camelContext.getManagementStrategy().getManagementAgent().setStatisticsLevel(ManagementStatisticsLevel.Extended);
+ return camelContext;
+ }
+
+ @Test
+ public void testSplitIsOneReply() {
+ assertEquals("done", template.requestBody("seda:split", "a,b,c"));
+ assertEquals("done", template.requestBody("seda:split", "d,e,f"));
+
+ assertEquals(2, hits("split", "seda://split", "out"));
+ }
+
+ @Test
+ public void testMulticastIsOneReply() {
+ assertEquals("done", template.requestBody("seda:multicast", "Hello
World"));
+
+ assertEquals(1, hits("multicast", "seda://multicast", "out"));
+ assertEquals(0, hits("sub", "direct://sub", "out"));
+ }
+
+ @Test
+ public void testRecipientListIsOneReply() {
+ assertEquals("done", template.requestBodyAndHeader("seda:list", "Hello
World", "to", "mock:b,direct:sub"));
+ assertEquals("done", template.requestBodyAndHeader("seda:list", "Bye
World", "to", "mock:b,direct:sub"));
+
+ assertEquals(2, hits("list", "seda://list", "out"));
+ assertEquals(0, hits("sub", "direct://sub", "out"));
+ }
+
+ @Test
+ public void testARouteCalledOverDirectSendsNoReply() {
+ for (int i = 0; i < 3; i++) {
+ template.requestBody("seda:call", "Hello " + i);
+ }
+
+ assertEquals(3, hits("call", "seda://call", "out"));
+ assertEquals(3, hits("call", "direct://sub", "out"));
+ assertEquals(0, hits("sub", "direct://sub", "out"));
+ }
+
+ private long hits(String routeId, String uri, String direction) {
+ return
context.getRuntimeEndpointRegistry().getEndpointStatistics().stream()
+ .filter(s -> routeId.equals(s.getRouteId()) &&
uri.equals(s.getUri()) && direction.equals(s.getDirection()))
+ .mapToLong(RuntimeEndpointRegistry.Statistic::getHits).sum();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:split").routeId("split").split(body().tokenize(",")).to("mock:split").end()
+ .transform(constant("done"));
+
from("seda:multicast").routeId("multicast").multicast().to("mock:a",
"direct:sub").end()
+ .transform(constant("done"));
+
from("seda:list").routeId("list").recipientList(header("to")).end().transform(constant("done"));
+ from("seda:call").routeId("call").to("direct:sub");
+ from("direct:sub").routeId("sub").to("mock:sub");
+ }
+ };
+ }
+}
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 b53a6ee48f2b..58fafe5f1b8f 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
@@ -285,7 +285,7 @@ before calling an endpoint). Before, such a recovery was
not counted. `ExchangeH
`true` when the error handler redelivered the exchange, without looking at the
header only. An exchange whose failure
is continued (`continued(true)`) no longer keeps its redelivery counter, as
with a handled failure.
-=== camel-management - the input hits of the runtime endpoint registry are the
messages a route received
+=== camel-management - the input and reply hits of the runtime endpoint
registry count each message once
The runtime endpoint registry (in extended statistics mode, as the
`RuntimeEndpointRegistry` MBean, the endpoints dev
console and the Camel CLI show it) now counts an input hit of a route when an
exchange enters the route through its
@@ -294,6 +294,10 @@ and the sub-exchanges of a Split, Multicast or Recipient
List were counted as in
route that consumes from `direct` (or got a message sent by a
`ProducerTemplate` to its `direct` endpoint) had no input
hits at all.
+Likewise, the reply of an InOut consumer is counted as an out hit of its
endpoint when the exchange is done in the route,
+once per message. Before, the completed copy of a Circuit Breaker and the
sub-exchanges of a Split, Multicast or
+Recipient List were counted as replies as well.
+
=== camel-core - a step after a redelivered step gets all its redeliveries
When a step of a route was redelivered and then succeeded, a later failure (of
another step of the route, or of a
