This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/27576-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit b7785c260b711568d8eea30865d5189b544cf7b8 Author: Guillaume Nodet <[email protected]> AuthorDate: Thu Oct 8 16:12:44 2026 +0000 [backport camel-4.22.x] CAMEL-24887: Fix route group MBean exchange counters to count once per exchange --- .../main/java/org/apache/camel/spi/UnitOfWork.java | 18 ++ .../camel/impl/engine/DefaultUnitOfWork.java | 21 +++ ...aultUnitOfWorkRouteStackLevelPredicateTest.java | 131 +++++++++++++++ .../camel/management/mbean/ManagedRouteGroup.java | 28 +++ .../ManagedRouteGroupExchangeCountTest.java | 187 +++++++++++++++++++++ .../ManagedRouteGroupFailuresHandledTest.java | 4 +- .../camel/management/ManagedRouteGroupTest.java | 6 +- .../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 7 + .../modules/ROOT/pages/route-group.adoc | 6 + 9 files changed, 403 insertions(+), 5 deletions(-) diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java b/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java index 99870ee8c594..1204c1c71609 100644 --- a/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java +++ b/core/camel-api/src/main/java/org/apache/camel/spi/UnitOfWork.java @@ -247,6 +247,24 @@ public interface UnitOfWork { */ int routeStackLevel(boolean includeRouteTemplate, boolean includeKamelet); + /** + * Gets the number of routes on the route stack that this {@link UnitOfWork} currently is being routed through, + * which match the given filter. + * <p/> + * For example to know whether the current route is the outermost route of a route group on the stack. + * <p/> + * The default returns {@code 0}, which means a custom {@link UnitOfWork} that does not override this method will + * not benefit from the route-group deduplication in {@code ManagedRouteGroup}. Implementations that maintain their + * own route stack should override this method. + * + * @param filter the filter + * @return the number of matching routes on the route stack + * @since 4.23 + */ + default int routeStackLevel(Predicate<Route> filter) { + return 0; + } + /** * Whether the unit of work should call the before/after process methods or not. */ diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java index 60ce4d4b8c3e..1035b1929bae 100644 --- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java +++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultUnitOfWork.java @@ -449,6 +449,27 @@ public class DefaultUnitOfWork implements UnitOfWork { } } + @Override + public int routeStackLevel(Predicate<Route> filter) { + lock.lock(); + try { + int level = 0; + if (route != null && filter.test(route)) { + level++; + } + if (routeStack != null) { + for (Route r : routeStack) { + if (filter.test(r)) { + level++; + } + } + } + return level; + } finally { + lock.unlock(); + } + } + private static int countRoute(Route r, boolean includeRouteTemplate, boolean includeKamelet) { if (r.isCreatedByKamelet()) { return includeKamelet ? 1 : 0; diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultUnitOfWorkRouteStackLevelPredicateTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultUnitOfWorkRouteStackLevelPredicateTest.java new file mode 100644 index 000000000000..34b4574e258a --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultUnitOfWorkRouteStackLevelPredicateTest.java @@ -0,0 +1,131 @@ +/* + * 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.impl.engine; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Route; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.spi.UnitOfWork; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit test for {@link DefaultUnitOfWork#routeStackLevel(java.util.function.Predicate)} (CAMEL-24887). + */ +public class DefaultUnitOfWorkRouteStackLevelPredicateTest extends ContextTestSupport { + + @Test + public void testEmptyStackReturnsZero() throws Exception { + UnitOfWork uow = new DefaultUnitOfWork(createExchangeWithBody("test")); + assertThat(uow.routeStackLevel(r -> true)).isZero(); + } + + @Test + public void testSingleRouteMatchingGroup() throws Exception { + UnitOfWork uow = new DefaultUnitOfWork(createExchangeWithBody("test")); + Route routeA = context.getRoute("a"); + + uow.pushRoute(routeA); + + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isEqualTo(1); + assertThat(uow.routeStackLevel(r -> "beta".equals(r.getGroup()))).isZero(); + } + + @Test + public void testNestedRoutesInSameGroup() throws Exception { + UnitOfWork uow = new DefaultUnitOfWork(createExchangeWithBody("test")); + Route routeA = context.getRoute("a"); + Route routeB = context.getRoute("b"); + + uow.pushRoute(routeA); + uow.pushRoute(routeB); + + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isEqualTo(2); + } + + @Test + public void testNestedRoutesDifferentGroups() throws Exception { + UnitOfWork uow = new DefaultUnitOfWork(createExchangeWithBody("test")); + Route routeA = context.getRoute("a"); + Route routeC = context.getRoute("c"); + + uow.pushRoute(routeA); + uow.pushRoute(routeC); + + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isEqualTo(1); + assertThat(uow.routeStackLevel(r -> "beta".equals(r.getGroup()))).isEqualTo(1); + } + + @Test + public void testPopReducesLevel() throws Exception { + UnitOfWork uow = new DefaultUnitOfWork(createExchangeWithBody("test")); + Route routeA = context.getRoute("a"); + Route routeB = context.getRoute("b"); + + uow.pushRoute(routeA); + uow.pushRoute(routeB); + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isEqualTo(2); + + uow.popRoute(); + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isEqualTo(1); + + uow.popRoute(); + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isZero(); + } + + @Test + public void testRouteWithNoGroupDoesNotMatch() throws Exception { + UnitOfWork uow = new DefaultUnitOfWork(createExchangeWithBody("test")); + Route routeD = context.getRoute("d"); + + uow.pushRoute(routeD); + + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isZero(); + assertThat(uow.routeStackLevel(r -> r.getGroup() == null)).isEqualTo(1); + } + + @Test + public void testMixedGroupsAndNoGroup() throws Exception { + UnitOfWork uow = new DefaultUnitOfWork(createExchangeWithBody("test")); + Route routeA = context.getRoute("a"); + Route routeD = context.getRoute("d"); + Route routeC = context.getRoute("c"); + + uow.pushRoute(routeA); + uow.pushRoute(routeD); + uow.pushRoute(routeC); + + assertThat(uow.routeStackLevel(r -> "alpha".equals(r.getGroup()))).isEqualTo(1); + assertThat(uow.routeStackLevel(r -> "beta".equals(r.getGroup()))).isEqualTo(1); + assertThat(uow.routeStackLevel(r -> r.getGroup() == null)).isEqualTo(1); + assertThat(uow.routeStackLevel(r -> true)).isEqualTo(3); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:a").routeId("a").routeGroup("alpha").to("mock:a"); + from("direct:b").routeId("b").routeGroup("alpha").to("mock:b"); + from("direct:c").routeId("c").routeGroup("beta").to("mock:c"); + from("direct:d").routeId("d").to("mock:d"); + } + }; + } +} diff --git a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java index fc42869151c0..b36d06eebd0f 100644 --- a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java +++ b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedRouteGroup.java @@ -20,6 +20,7 @@ import java.util.Date; import java.util.List; import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; import org.apache.camel.ManagementStatisticsLevel; import org.apache.camel.Route; import org.apache.camel.ServiceStatus; @@ -27,6 +28,7 @@ import org.apache.camel.api.management.ManagedResource; import org.apache.camel.api.management.mbean.ManagedRouteGroupMBean; import org.apache.camel.api.management.mbean.RouteError; import org.apache.camel.spi.ManagementStrategy; +import org.apache.camel.spi.UnitOfWork; import org.apache.camel.util.TimeUtils; @ManagedResource(description = "Managed Route Group") @@ -207,6 +209,32 @@ public class ManagedRouteGroup extends ManagedPerformanceCounter implements Mana } } + @Override + public void processExchange(Exchange exchange, String type) { + if (isOutermostRouteOfGroup(exchange)) { + super.processExchange(exchange, type); + } + } + + @Override + public void completedExchange(Exchange exchange, long time) { + if (isOutermostRouteOfGroup(exchange)) { + super.completedExchange(exchange, time); + } + } + + @Override + public void failedExchange(Exchange exchange) { + if (isOutermostRouteOfGroup(exchange)) { + super.failedExchange(exchange); + } + } + + private boolean isOutermostRouteOfGroup(Exchange exchange) { + UnitOfWork uow = exchange.getUnitOfWork(); + return uow == null || uow.routeStackLevel(r -> group.equals(r.getGroup())) <= 1; + } + @Override public void onTimer() { super.onTimer(); diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupExchangeCountTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupExchangeCountTest.java new file mode 100644 index 000000000000..531420a4792c --- /dev/null +++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupExchangeCountTest.java @@ -0,0 +1,187 @@ +/* + * 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 java.util.concurrent.TimeUnit; + +import org.apache.camel.api.management.ManagedCamelContext; +import org.apache.camel.api.management.mbean.ManagedRouteGroupMBean; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledOnOs; +import org.junit.jupiter.api.condition.OS; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Tests that a route group counts each exchange once per entry into the group: calls nested inside another route of the + * same group are not counted again (CAMEL-24887). + */ +@DisabledOnOs(OS.AIX) +public class ManagedRouteGroupExchangeCountTest extends ManagementTestSupport { + + @Test + public void testGroupCountsExchangeOnce() throws Exception { + ManagedCamelContext mcc = context.getCamelContextExtension().getContextPlugin(ManagedCamelContext.class); + + // send 1 message through x1 -> x2 (both in group "g") + template.sendBody("direct:x1", "Hello"); + + ManagedRouteGroupMBean group = mcc.getManagedRouteGroup("g"); + assertNotNull(group); + assertEquals(1, group.getExchangesCompleted()); + assertEquals(1, group.getExchangesTotal()); + assertEquals(0, group.getExchangesInflight()); + + // send 2 more messages + template.sendBody("direct:x1", "World"); + template.sendBody("direct:x1", "!"); + + assertEquals(3, group.getExchangesCompleted()); + assertEquals(3, group.getExchangesTotal()); + } + + @Test + public void testGroupCountsOnceWhenRoutingThroughNonGroupRoute() throws Exception { + ManagedCamelContext mcc = context.getCamelContextExtension().getContextPlugin(ManagedCamelContext.class); + + // send 1 message through x1 (group "g") -> y (no group) -> x2 (group "g") + template.sendBody("direct:x1-via-y", "Hello"); + + ManagedRouteGroupMBean group = mcc.getManagedRouteGroup("g"); + assertNotNull(group); + assertEquals(1, group.getExchangesCompleted()); + assertEquals(1, group.getExchangesTotal()); + } + + @Test + public void testGroupCountsWhenEntryIsOutsideGroup() throws Exception { + ManagedCamelContext mcc = context.getCamelContextExtension().getContextPlugin(ManagedCamelContext.class); + + // send 1 message to a route outside the group that calls x2 (in group "g") directly + template.sendBody("direct:outside", "Hello"); + + ManagedRouteGroupMBean group = mcc.getManagedRouteGroup("g"); + assertNotNull(group); + assertEquals(1, group.getExchangesCompleted()); + } + + @Test + public void testGroupCountsFailure() throws Exception { + ManagedCamelContext mcc = context.getCamelContextExtension().getContextPlugin(ManagedCamelContext.class); + + // send 1 message that will fail in x2-fail (group "g") + try { + template.sendBody("direct:x1-fail", "Hello"); + } catch (Exception e) { + // expected + } + + ManagedRouteGroupMBean group = mcc.getManagedRouteGroup("g"); + assertNotNull(group); + assertEquals(1, group.getExchangesFailed()); + assertEquals(0, group.getExchangesCompleted()); + assertEquals(0, group.getExchangesInflight()); + } + + @Test + public void testTwoGroupsOnOnePath() throws Exception { + ManagedCamelContext mcc = context.getCamelContextExtension().getContextPlugin(ManagedCamelContext.class); + + // send 1 message through g1-route (group "g1") -> g2-route (group "g2") + template.sendBody("direct:g1-route", "Hello"); + + ManagedRouteGroupMBean group1 = mcc.getManagedRouteGroup("g1"); + ManagedRouteGroupMBean group2 = mcc.getManagedRouteGroup("g2"); + assertNotNull(group1); + assertNotNull(group2); + assertEquals(1, group1.getExchangesCompleted()); + assertEquals(1, group2.getExchangesCompleted()); + } + + @Test + public void testWireTapGroupCounts() throws Exception { + ManagedCamelContext mcc = context.getCamelContextExtension().getContextPlugin(ManagedCamelContext.class); + + getMockEndpoint("mock:wt-done").expectedMessageCount(1); + + template.sendBody("direct:wt", "Hello"); + + assertMockEndpointsSatisfied(); + + // wait for the wireTapped exchange to complete in route "wt-target" + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1L, mcc.getManagedRoute("wt-target").getExchangesCompleted())); + + // each route counts independently at the route level + assertEquals(1, mcc.getManagedRoute("wt").getExchangesCompleted()); + assertEquals(1, mcc.getManagedRoute("wt-target").getExchangesCompleted()); + + // the group count is consistent with how the CamelContext MBean counts exchanges + long ctxCompleted = mcc.getManagedCamelContext().getExchangesCompleted(); + + ManagedRouteGroupMBean group = mcc.getManagedRouteGroup("g"); + assertNotNull(group); + assertEquals(ctxCompleted, group.getExchangesCompleted()); + assertEquals(ctxCompleted, group.getExchangesTotal()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + // x1 -> x2, both in group "g" + from("direct:x1").routeId("x1").routeGroup("g") + .to("direct:x2"); + from("direct:x2").routeId("x2").routeGroup("g") + .log("x2 done"); + + // x1 -> y (no group) -> x2, tests routing through non-group route + from("direct:x1-via-y").routeId("x1-via-y").routeGroup("g") + .to("direct:y"); + from("direct:y").routeId("y") + .to("direct:x2"); + + // outside (no group) -> x2 (group "g") + from("direct:outside").routeId("outside") + .to("direct:x2"); + + // failure path: x1-fail -> x2-fail (throws), both in group "g" + from("direct:x1-fail").routeId("x1-fail").routeGroup("g") + .to("direct:x2-fail"); + from("direct:x2-fail").routeId("x2-fail").routeGroup("g") + .throwException(new RuntimeException("Test failure")); + + // two groups: g1-route (group "g1") -> g2-route (group "g2") + from("direct:g1-route").routeId("g1-route").routeGroup("g1") + .to("direct:g2-route"); + from("direct:g2-route").routeId("g2-route").routeGroup("g2") + .log("g2 done"); + + // wireTap: wt (group "g") wireTaps to wt-target (group "g") + from("direct:wt").routeId("wt").routeGroup("g") + .wireTap("direct:wt-target"); + from("direct:wt-target").routeId("wt-target").routeGroup("g") + .to("mock:wt-done"); + } + }; + } + +} diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupFailuresHandledTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupFailuresHandledTest.java index 38b7a53e1064..5e4127863ccb 100644 --- a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupFailuresHandledTest.java +++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupFailuresHandledTest.java @@ -42,8 +42,8 @@ public class ManagedRouteGroupFailuresHandledTest extends ManagementTestSupport ManagedRouteGroupMBean group = mcc.getManagedRouteGroup("flow"); assertNotNull(group); - // group stats must aggregate across all member routes: trigger, step1 and step2 each completed once - assertEquals(3, group.getExchangesCompleted()); + // group counts each exchange once (outermost route of the group) + assertEquals(1, group.getExchangesCompleted()); // the handled failure recorded on the trigger route must be reflected at the group level assertEquals(1, group.getFailuresHandled()); diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupTest.java index aee8516341c8..bdab8fb39e6f 100644 --- a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupTest.java +++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteGroupTest.java @@ -56,11 +56,11 @@ public class ManagedRouteGroupTest extends ManagementTestSupport { Integer size = (Integer) mbeanServer.getAttribute(on, "GroupSize"); if ("first".equals(group)) { assertEquals(3, size); - // group stats aggregate across all member routes (start, a, e each processed the exchange) - assertEquals(3, val); + // start -> a and start -> e are nested: start is outermost, so counted once + assertEquals(1, val); } else { assertEquals(2, size); - // group stats aggregate across all member routes (c, d each processed the exchange) + // c and d are called sequentially by "start" (not nested), each is outermost for "second" assertEquals(2, val); } diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index 8c2ad20d563e..9eca5cfbffa4 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -13,6 +13,13 @@ See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details. == Upgrading from 4.22.1 to 4.22.2 +=== camel-management - Route group exchange counters count once per entry into the group again + +In 4.22.1, the route group MBean (`ManagedRouteGroupMBean`) exchange counters and processing times +counted each exchange once per member route instead of once per entry into the group. This was an +unintended side effect of CAMEL-24590. The counters now count each exchange once per entry into the +group again, as in 4.22.0: calls nested inside another route of the same group are not counted again. + === camel-docling - output path handling The `CamelDoclingOutputFilePath` header, which selects the CLI output directory, is now normalized lexically diff --git a/docs/user-manual/modules/ROOT/pages/route-group.adoc b/docs/user-manual/modules/ROOT/pages/route-group.adoc index 3e070cc84cd0..8157a157bfe8 100644 --- a/docs/user-manual/modules/ROOT/pages/route-group.adoc +++ b/docs/user-manual/modules/ROOT/pages/route-group.adoc @@ -177,6 +177,12 @@ Camel has optional support for xref:jmx.adoc[JMX management]. This includes mana The following code will get the JMX MBean `ManagedRouteGroupMBean` for the route group with name `order`. Using this MBean you can get metrics, and also operations to start and stop all routes in this group. +Each exchange is counted once per entry into the group: calls nested inside another route of the +same group are not counted again. For example, if route A (group G) calls route B (group G) via +`direct:`, the group counts it once. However, if an outside route calls A and then B sequentially, +each call is a separate entry, so the group counts two. +The per-route statistics are available on each route's `ManagedRouteMBean`. + For example to get the total number of `Exchange`(s) that has failed within this group: [source,java]
