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 eb341770d40a CAMEL-26780: camel-core - the runtime endpoint registry
counts an input hit when an exchange enters the route (#29476)
eb341770d40a is described below
commit eb341770d40a61f975449c751cd23aff48584caa
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 9 15:05:41 2026 +0100
CAMEL-26780: camel-core - the runtime endpoint registry counts an input hit
when an exchange enters the route (#29476)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../resilience4j/ResilienceInputHitsTest.java | 81 +++++++++++++++
.../engine/DefaultRuntimeEndpointRegistry.java | 81 +++++++++------
.../management/ManagedEndpointInputHitsTest.java | 110 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 ++
4 files changed, 253 insertions(+), 28 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
new file mode 100644
index 000000000000..2f47bd2129ac
--- /dev/null
+++
b/components/camel-resilience4j/src/test/java/org/apache/camel/component/resilience4j/ResilienceInputHitsTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.component.resilience4j;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ManagementStatisticsLevel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.RuntimeEndpointRegistry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+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.
+ */
+public class ResilienceInputHitsTest extends CamelTestSupport {
+
+ @Override
+ protected boolean useJmx() {
+ return true;
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+
context.getManagementStrategy().getManagementAgent().setStatisticsLevel(ManagementStatisticsLevel.Extended);
+ return context;
+ }
+
+ @Test
+ public void testOneInputPerMessage() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("in stock", "in
stock", "in stock");
+ for (int i = 0; i < 3; i++) {
+ template.sendBody("seda:check", "CAMEL-MUG");
+ }
+ MockEndpoint.assertIsSatisfied(context);
+
+ assertEquals(3, hits("check", "seda://check", "in"));
+ assertEquals(3, hits("supplier", "direct://supplier", "in"));
+ }
+
+ 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:check").routeId("check")
+ .circuitBreaker()
+ .to("direct:supplier")
+ .onFallback()
+ .transform().constant("not called")
+ .end()
+ .to("mock:result");
+
+
from("direct:supplier").routeId("supplier").transform().constant("in stock");
+ }
+ };
+ }
+}
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 0fc53cfee529..83e840dc5c91 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
@@ -28,10 +28,10 @@ import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Message;
+import org.apache.camel.Route;
import org.apache.camel.StreamCache;
import org.apache.camel.spi.CamelEvent;
import org.apache.camel.spi.CamelEvent.ExchangeCompletedEvent;
-import org.apache.camel.spi.CamelEvent.ExchangeCreatedEvent;
import org.apache.camel.spi.CamelEvent.ExchangeFailedEvent;
import org.apache.camel.spi.CamelEvent.ExchangeSendingEvent;
import org.apache.camel.spi.CamelEvent.ExchangeSentEvent;
@@ -45,6 +45,7 @@ import org.apache.camel.support.EndpointSizeStatistics;
import org.apache.camel.support.EventNotifierSupport;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.LRUCacheFactory;
+import org.apache.camel.support.RoutePolicySupport;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -65,6 +66,7 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
private EndpointUtilizationStatistics outputUtilization;
private EndpointSizeStatistics inputSizeStats;
private EndpointSizeStatistics outputSizeStats;
+ private final RoutePolicySupport inputHits = new InputHits();
@Override
public boolean isEnabled() {
@@ -212,6 +214,13 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
outputSizeStats = new EndpointSizeStatistics(limit);
}
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);
LOG.debug(
"Runtime endpoint registry is in extended mode gathering
usage statistics of all incoming and outgoing endpoints (cache limit: {})",
limit);
@@ -285,31 +294,6 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
}
}
}
- } else if (extended && event instanceof ExchangeCreatedEvent ece) {
- // we only capture details in extended mode
- Exchange exchange = ece.getExchange();
- Endpoint endpoint = exchange.getFromEndpoint();
- if (endpoint != null) {
- String routeId = exchange.getFromRouteId();
- String uri = endpoint.getEndpointUri();
- String key = asUtilizationKey(routeId, uri);
- if (key != null) {
- inputUtilization.onHit(key);
- if (messageSizeEnabled) {
- Message message = exchange.getIn();
- MessageSizeStrategy strategy =
getCamelContext().getMessageSizeStrategy();
- long bodySize = strategy.computeBodySize(message);
- long headersSize =
strategy.computeHeadersSize(message);
- inputSizeStats.onHit(key, bodySize, headersSize);
-
exchange.setProperty(ExchangePropertyKey.MESSAGE_BODY_SIZE, bodySize);
-
exchange.setProperty(ExchangePropertyKey.MESSAGE_HEADERS_SIZE, headersSize);
- // reset stream cache so the body is re-readable
during routing
- if (message.getBody() instanceof StreamCache sc) {
- sc.reset();
- }
- }
- }
- }
} else if (event instanceof ExchangeSendingEvent ese) {
Endpoint endpoint = ese.getEndpoint();
Exchange exchange = ese.getExchange();
@@ -411,8 +395,7 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
@Override
public boolean isEnabled(CamelEvent event) {
- return enabled && event instanceof ExchangeCreatedEvent
- || event instanceof ExchangeSendingEvent
+ return event instanceof ExchangeSendingEvent
|| event instanceof ExchangeSentEvent
|| event instanceof ExchangeCompletedEvent
|| event instanceof ExchangeFailedEvent
@@ -430,6 +413,48 @@ public class DefaultRuntimeEndpointRegistry extends
EventNotifierSupport impleme
return null;
}
+ /**
+ * Counts an input hit of the route the exchange enters, under the URI of
the route's consumer (which differs from
+ * the endpoint of the route when a component delegates to another
consumer, as rest-openapi does to platform-http).
+ */
+ private void onRouteInput(Route route, Exchange exchange) {
+ Endpoint endpoint = route.getConsumer() != null ?
route.getConsumer().getEndpoint() : route.getEndpoint();
+ String key = endpoint != null ? asUtilizationKey(route.getRouteId(),
endpoint.getEndpointUri()) : null;
+ if (key == null || inputUtilization == null) {
+ return;
+ }
+ inputUtilization.onHit(key);
+ if (messageSizeEnabled) {
+ Message message = exchange.getIn();
+ MessageSizeStrategy strategy =
getCamelContext().getMessageSizeStrategy();
+ long bodySize = strategy.computeBodySize(message);
+ long headersSize = strategy.computeHeadersSize(message);
+ inputSizeStats.onHit(key, bodySize, headersSize);
+ exchange.setProperty(ExchangePropertyKey.MESSAGE_BODY_SIZE,
bodySize);
+ exchange.setProperty(ExchangePropertyKey.MESSAGE_HEADERS_SIZE,
headersSize);
+ // reset stream cache so the body is re-readable during routing
+ 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. */
+ private final class InputHits extends RoutePolicySupport {
+
+ @Override
+ public void onExchangeBegin(Route route, Exchange exchange) {
+ if (extended && enabled) {
+ onRouteInput(route, exchange);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "RuntimeEndpointRegistryInputHits";
+ }
+ }
+
private static String asUtilizationKey(String routeId, String uri) {
if (routeId == null || uri == null) {
return null;
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedEndpointInputHitsTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedEndpointInputHitsTest.java
new file mode 100644
index 000000000000..f17da64fe0ee
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedEndpointInputHitsTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.component.mock.MockEndpoint;
+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;
+
+/**
+ * The input hits of a route are the messages its consumer received: the
sub-exchanges of a split, multicast or
+ * recipient list are no input of the route, and a route that consumes from
direct has an input hit per message it gets.
+ */
+@DisabledOnOs(OS.AIX)
+public class ManagedEndpointInputHitsTest extends ManagementTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+
camelContext.getManagementStrategy().getManagementAgent().setStatisticsLevel(ManagementStatisticsLevel.Extended);
+ return camelContext;
+ }
+
+ @Test
+ public void testSplitIsOneInput() throws Exception {
+ getMockEndpoint("mock:split").expectedMessageCount(6);
+ template.sendBody("seda:split", "a,b,c");
+ template.sendBody("seda:split", "d,e,f");
+ MockEndpoint.assertIsSatisfied(context);
+
+ assertEquals(2, hits("split", "seda://split", "in"));
+ }
+
+ @Test
+ public void testMulticastIsOneInputAndAnInputOfTheRouteItCallsOverDirect()
throws Exception {
+ getMockEndpoint("mock:a").expectedMessageCount(1);
+ getMockEndpoint("mock:sub").expectedMessageCount(1);
+ template.sendBody("seda:multicast", "Hello World");
+ MockEndpoint.assertIsSatisfied(context);
+
+ assertEquals(1, hits("multicast", "seda://multicast", "in"));
+ assertEquals(1, hits("multicast", "direct://sub", "out"));
+ assertEquals(1, hits("sub", "direct://sub", "in"));
+ }
+
+ @Test
+ public void
testRecipientListIsOneInputAndAnInputOfTheRouteItCallsOverDirect() throws
Exception {
+ getMockEndpoint("mock:b").expectedMessageCount(2);
+ getMockEndpoint("mock:sub").expectedMessageCount(2);
+ template.sendBodyAndHeader("seda:list", "Hello World", "to",
"mock:b,direct:sub");
+ template.sendBodyAndHeader("seda:list", "Bye World", "to",
"mock:b,direct:sub");
+ MockEndpoint.assertIsSatisfied(context);
+
+ assertEquals(2, hits("list", "seda://list", "in"));
+ assertEquals(2, hits("sub", "direct://sub", "in"));
+ }
+
+ @Test
+ public void testDirectIsAnInputOfTheRouteThatConsumesIt() throws Exception
{
+ getMockEndpoint("mock:sub").expectedMessageCount(3);
+ for (int i = 0; i < 3; i++) {
+ template.sendBody("seda:call", "Hello " + i);
+ }
+ MockEndpoint.assertIsSatisfied(context);
+
+ assertEquals(3, hits("call", "seda://call", "in"));
+ assertEquals(3, hits("call", "direct://sub", "out"));
+ assertEquals(3, hits("sub", "direct://sub", "in"));
+ }
+
+ 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");
+
from("seda:multicast").routeId("multicast").multicast().to("mock:a",
"direct:sub");
+ from("seda:list").routeId("list").recipientList(header("to"));
+ 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 3f82626e4ff3..3bd88a9ea434 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
@@ -273,6 +273,15 @@ 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
+
+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
+consumer, as the route counts its exchanges. Before, it counted every created
exchange: the copy of a Circuit Breaker
+and the sub-exchanges of a Split, Multicast or Recipient List were counted as
input hits of the route as well, and a
+route that consumes from `direct` (or got a message sent by a
`ProducerTemplate` to its `direct` endpoint) had no input
+hits at all.
+
=== 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
