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 648434b050c0 CAMEL-24978: camel-core - Internal processor advices: fix
bugs found in a deep review (#26811)
648434b050c0 is described below
commit 648434b050c0c35a952fa6bd8c87c2bb7ce17d1b
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Sep 24 11:40:37 2026 +0200
CAMEL-24978: camel-core - Internal processor advices: fix bugs found in a
deep review (#26811)
A deep review of the routing engine's internal processor
(CamelInternalProcessor, SharedCamelInternalProcessor, AdviceIterator and the
built-in advices) found the bugs below, each with a test that fails without the
fix. The normal path, where each before is followed by its after, is sound; all
of these are on edge paths.
After advices were not run on the debugger's skip over, or when an advice
failed in before. On skip over all the before advices had run but the branch
only called the callback, so the JMX inflight counter of the skipped processor
never went down, the message history entry was left unfinished, and with
tracing on the exchange stayed in the tracer's event notifier map. When a
before failed, the advices that had already run were never undone, so the
route's inflight count stayed up, the [...]
rest-openapi and rest-postman ran each operation twice after a route
restart: the consumer is created again on restart and another processor advice
was added each time without removing the previous one, so every restart added
one more invocation per request. The advice from the previous start is now
removed first.
Closes #26811
---
.../vertx/PlatformHttpRestOpenApiConsumerTest.java | 40 ++++++
.../vertx/PlatformHttpRestPostmanConsumerTest.java | 40 ++++++
.../rest/openapi/RestOpenApiEndpoint.java | 5 +
.../rest/postman/RestPostmanEndpoint.java | 5 +
.../apache/camel/impl/engine/AdviceIterator.java | 29 +++-
.../camel/impl/engine/CamelInternalProcessor.java | 75 ++++------
.../engine/CamelInternalProcessorAdviceTest.java | 154 +++++++++++++++++++++
.../BacklogTracerAggregateStandbyTest.java | 74 ++++++++++
.../management/ManagedProcessorSkipOverTest.java | 64 +++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 9 ++
10 files changed, 442 insertions(+), 53 deletions(-)
diff --git
a/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestOpenApiConsumerTest.java
b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestOpenApiConsumerTest.java
index fb7cca2564d5..14e03ca2f7f8 100644
---
a/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestOpenApiConsumerTest.java
+++
b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestOpenApiConsumerTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.platform.http.vertx;
+import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
@@ -24,6 +26,7 @@ import org.junit.jupiter.api.Test;
import static io.restassured.RestAssured.given;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.equalToCompressingWhiteSpace;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class PlatformHttpRestOpenApiConsumerTest {
@@ -270,4 +273,41 @@ public class PlatformHttpRestOpenApiConsumerTest {
}
}
+ @Test
+ public void testRestOpenApiRouteRestart() throws Exception {
+ final CamelContext context =
VertxPlatformHttpEngineTest.createCamelContext();
+ final AtomicInteger counter = new AtomicInteger();
+
+ try {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("rest-openapi:classpath:openapi-v3.json?missingOperation=ignore").routeId("api")
+ .to("mock:result");
+
+ from("direct:getPetById")
+ .process(e -> counter.incrementAndGet())
+ .setBody().constant("{\"pet\": \"tony the
tiger\"}");
+ }
+ });
+
+ VertxPlatformHttpEngineTest.startCamelContext(context);
+
+ // restart the route, which creates the consumer again
+ context.getRouteController().stopRoute("api");
+ context.getRouteController().startRoute("api");
+
+ given()
+ .when()
+ .get("/api/v3/pet/123")
+ .then()
+ .statusCode(200)
+ .body(equalTo("{\"pet\": \"tony the tiger\"}"));
+
+ // the operation must only be invoked once
+ assertEquals(1, counter.get());
+ } finally {
+ context.stop();
+ }
+ }
}
diff --git
a/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestPostmanConsumerTest.java
b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestPostmanConsumerTest.java
index ae2f889441b5..c3d004f03861 100644
---
a/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestPostmanConsumerTest.java
+++
b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/PlatformHttpRestPostmanConsumerTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.platform.http.vertx;
+import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
@@ -258,4 +260,42 @@ class PlatformHttpRestPostmanConsumerTest {
context.stop();
}
}
+
+ @Test
+ void shouldInvokeRequestOnceAfterRouteRestart() throws Exception {
+ final CamelContext context =
VertxPlatformHttpEngineTest.createCamelContext();
+ final AtomicInteger counter = new AtomicInteger();
+
+ try {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("rest-postman:classpath:postman-petstore.json?missingRequest=ignore").routeId("api")
+ .to("mock:result");
+
+ from("direct:getPetById")
+ .process(e -> counter.incrementAndGet())
+ .setBody().constant("{\"pet\": \"tony the
tiger\"}");
+ }
+ });
+
+ VertxPlatformHttpEngineTest.startCamelContext(context);
+
+ // restart the route, which creates the consumer again
+ context.getRouteController().stopRoute("api");
+ context.getRouteController().startRoute("api");
+
+ given()
+ .when()
+ .get("/api/v3/pet/123")
+ .then()
+ .statusCode(200)
+ .body(equalTo("{\"pet\": \"tony the tiger\"}"));
+
+ // the operation must only be invoked once
+ assertThat(counter.get()).isEqualTo(1);
+ } finally {
+ context.stop();
+ }
+ }
}
diff --git
a/components/camel-rest-openapi/src/main/java/org/apache/camel/component/rest/openapi/RestOpenApiEndpoint.java
b/components/camel-rest-openapi/src/main/java/org/apache/camel/component/rest/openapi/RestOpenApiEndpoint.java
index 1b108d15b081..c343e4fe5660 100644
---
a/components/camel-rest-openapi/src/main/java/org/apache/camel/component/rest/openapi/RestOpenApiEndpoint.java
+++
b/components/camel-rest-openapi/src/main/java/org/apache/camel/component/rest/openapi/RestOpenApiEndpoint.java
@@ -240,6 +240,11 @@ public final class RestOpenApiEndpoint extends
DefaultEndpoint {
if (advice != null) {
ip.removeAdvice(advice);
}
+ // remove the advice from a previous start of the route, as the
consumer is created again when restarted
+ RestOpenApiProcessorAdvice existing =
ip.getAdvice(RestOpenApiProcessorAdvice.class);
+ if (existing != null) {
+ ip.removeAdvice(existing);
+ }
ip.addAdvice(new RestOpenApiProcessorAdvice(openApiProcessor));
}
diff --git
a/components/camel-rest-postman/src/main/java/org/apache/camel/component/rest/postman/RestPostmanEndpoint.java
b/components/camel-rest-postman/src/main/java/org/apache/camel/component/rest/postman/RestPostmanEndpoint.java
index f7e0da531bb4..0f54879cbab6 100644
---
a/components/camel-rest-postman/src/main/java/org/apache/camel/component/rest/postman/RestPostmanEndpoint.java
+++
b/components/camel-rest-postman/src/main/java/org/apache/camel/component/rest/postman/RestPostmanEndpoint.java
@@ -209,6 +209,11 @@ public class RestPostmanEndpoint extends DefaultEndpoint {
if (advice != null) {
ip.removeAdvice(advice);
}
+ // remove the advice from a previous start of the route, as the
consumer is created again when restarted
+ RestPostmanProcessorAdvice existing =
ip.getAdvice(RestPostmanProcessorAdvice.class);
+ if (existing != null) {
+ ip.removeAdvice(existing);
+ }
ip.addAdvice(new RestPostmanProcessorAdvice(restPostmanProcessor));
}
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AdviceIterator.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AdviceIterator.java
index d89fdcc92f32..8002a5f0ec24 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AdviceIterator.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AdviceIterator.java
@@ -28,9 +28,24 @@ final class AdviceIterator {
}
static void runAfterTasks(List<? extends CamelInternalProcessorAdvice>
advices, Object[] states, Exchange exchange) {
- int stateIndex = states.length - 1;
+ runAfterTasks(advices, advices.size(), states, states.length,
exchange);
+ }
+
+ /**
+ * Runs the after of the first advices in reverse order.
+ *
+ * @param advices the advices
+ * @param count number of advices (from the start) to run after for,
such as those whose before was run
+ * @param states the states
+ * @param stateCount number of states (from the start) that belongs to
these advices
+ * @param exchange the exchange
+ */
+ static void runAfterTasks(
+ List<? extends CamelInternalProcessorAdvice> advices, int count,
Object[] states, int stateCount,
+ Exchange exchange) {
+ int stateIndex = stateCount - 1;
- for (int i = advices.size() - 1; i >= 0; i--) {
+ for (int i = count - 1; i >= 0; i--) {
CamelInternalProcessorAdvice task = advices.get(i);
Object state = null;
if (task.hasState()) {
@@ -44,8 +59,14 @@ final class AdviceIterator {
try {
task.after(exchange, state);
} catch (Exception e) {
- exchange.setException(e);
- // allow all advices to complete even if there was an exception
+ // allow all advices to complete even if there was an exception,
+ // and do not lose the exception the exchange already failed with
+ Exception existing = exchange.getException();
+ if (existing == null) {
+ exchange.setException(e);
+ } else if (existing != e) {
+ existing.addSuppressed(e);
+ }
}
}
}
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
index 4a5400cb3d57..fb4088bad17f 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
@@ -194,7 +194,7 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
@Override
public void addRouteInflightRepositoryAdvice(InflightRepository
inflightRepository, String routeId) {
- addAdvice(new
CamelInternalProcessor.RouteInflightRepositoryAdvice(camelContext.getInflightRepository(),
routeId));
+ addAdvice(new
CamelInternalProcessor.RouteInflightRepositoryAdvice(inflightRepository,
routeId));
}
@Override
@@ -337,7 +337,7 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
states[j++] = state;
}
} catch (Exception e) {
- return handleException(exchange, originalCallback, e,
afterTask);
+ return handleException(exchange, originalCallback, e,
afterTask, i, j);
}
}
@@ -353,7 +353,8 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
last.setDebugSkipOver(true);
}
// skip because the processor is specially disabled (such as from
debugger)
- originalCallback.done(true);
+ // the before advices have been executed, so the after advices
must be executed as well
+ afterTask.done(true);
return true;
}
@@ -444,9 +445,21 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
}
private boolean handleException(
- Exchange exchange, AsyncCallback originalCallback, Exception e,
CamelInternalTask afterTask) {
+ Exchange exchange, AsyncCallback originalCallback, Exception e,
CamelInternalTask afterTask,
+ int count, int stateCount) {
// error in before so break out
exchange.setException(e);
+ try {
+ // the advices whose before was executed must have their after
executed as well
+ // (such as to remove from inflight repository, and done the unit
of work)
+ AdviceIterator.runAfterTasks(advices, count,
afterTask.getStates(), stateCount, exchange);
+ } finally {
+ handleExceptionDone(originalCallback, afterTask);
+ }
+ return true;
+ }
+
+ private void handleExceptionDone(AsyncCallback originalCallback,
CamelInternalTask afterTask) {
try {
originalCallback.done(true);
} finally {
@@ -455,7 +468,6 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
taskFactory.release(afterTask);
}
}
- return true;
}
@Override
@@ -711,9 +723,9 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
input.getShortName(), input.getLabel(),
level, exchangeId, correlationExchangeId,
breadcrumbId, rest, template, data);
if (exchange.getFromEndpoint() instanceof
EndpointServiceLocation esl) {
- first.setEndpointServiceUrl(esl.getServiceUrl());
- first.setEndpointServiceProtocol(esl.getServiceProtocol());
- first.setEndpointServiceMetadata(esl.getServiceMetadata());
+ last.setEndpointServiceUrl(esl.getServiceUrl());
+ last.setEndpointServiceProtocol(esl.getServiceProtocol());
+ last.setEndpointServiceMetadata(esl.getServiceMetadata());
}
backlogTracer.traceEvent(last);
doneProcessing(exchange, last);
@@ -784,6 +796,9 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
@Override
public DefaultBacklogTracerEventMessage before(Exchange exchange)
throws Exception {
+ if (!backlogTracer.shouldTrace(processorDefinition, exchange)) {
+ return null;
+ }
String exchangeId = exchange.getExchangeId();
String correlationExchangeId =
exchange.getProperty(ExchangePropertyKey.CORRELATION_ID, String.class);
String breadcrumbId =
exchange.getIn().getHeader(Exchange.BREADCRUMB_ID, String.class);
@@ -838,9 +853,9 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
processorDefinition.getShortName(),
processorDefinition.getLabel(),
level, exchangeId, correlationExchangeId,
breadcrumbId, false, false, data);
if (exchange.getFromEndpoint() instanceof
EndpointServiceLocation esl) {
- first.setEndpointServiceUrl(esl.getServiceUrl());
- first.setEndpointServiceProtocol(esl.getServiceProtocol());
- first.setEndpointServiceMetadata(esl.getServiceMetadata());
+ last.setEndpointServiceUrl(esl.getServiceUrl());
+ last.setEndpointServiceProtocol(esl.getServiceProtocol());
+ last.setEndpointServiceMetadata(esl.getServiceMetadata());
}
backlogTracer.traceEvent(last);
doneProcessing(exchange, last);
@@ -973,44 +988,6 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
return null;
}
- private SynchronizationAdapter createAggregateOnCompletion(
- String source, DefaultBacklogTracerEventMessage pseudoFirst) {
- return new SynchronizationAdapter() {
- @Override
- public void onDone(Exchange exchange) {
- // create pseudo last for the aggregate
- String routeId = routeDefinition != null ?
routeDefinition.getRouteId() : null;
- String fromRouteId = exchange.getFromRouteId();
- String exchangeId = exchange.getExchangeId();
- String correlationExchangeId =
exchange.getProperty(ExchangePropertyKey.CORRELATION_ID, String.class);
- String breadcrumbId =
exchange.getIn().getHeader(Exchange.BREADCRUMB_ID, String.class);
- boolean includeExchangeProperties =
backlogTracer.isIncludeExchangeProperties();
- boolean includeExchangeVariables =
backlogTracer.isIncludeExchangeVariables();
- long created = exchange.getClock().getCreated();
- int level = pseudoFirst.getToNodeLevel();
- String toNode = pseudoFirst.getToNode();
- String toNodeShortName = pseudoFirst.getToNodeShortName();
- String toNodeLabel = pseudoFirst.getToNodeLabel();
- JsonObject data =
MessageHelper.dumpAsJSonObject(exchange.getIn(), includeExchangeProperties,
- includeExchangeVariables, true,
- true, backlogTracer.isBodyIncludeStreams(),
backlogTracer.isBodyIncludeFiles(),
- backlogTracer.getBodyMaxChars());
- DefaultBacklogTracerEventMessage pseudoLast = new
DefaultBacklogTracerEventMessage(
- camelContext,
- false, true,
backlogTracer.incrementTraceCounter(), created, source, fromRouteId, routeId,
toNode,
- null, null,
- null, toNodeShortName, toNodeLabel,
- level, exchangeId, correlationExchangeId,
breadcrumbId, rest, template, data);
- backlogTracer.traceEvent(pseudoLast);
- doneProcessing(exchange, pseudoLast);
- doneProcessing(exchange, pseudoFirst);
- // to not be confused then lets store duration on
first/last as (first = 0, last = total time to process)
- pseudoLast.setElapsed(pseudoFirst.getElapsed());
- pseudoFirst.setElapsed(0);
- }
- };
- }
-
@Override
public void after(Exchange exchange, DefaultBacklogTracerEventMessage
data) throws Exception {
if (data != null) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/engine/CamelInternalProcessorAdviceTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/CamelInternalProcessorAdviceTest.java
new file mode 100644
index 000000000000..f7b24a21c2ec
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/CamelInternalProcessorAdviceTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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 java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.CamelInternalProcessorAdvice;
+import org.apache.camel.spi.InternalProcessor;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+public class CamelInternalProcessorAdviceTest extends ContextTestSupport {
+
+ private final AtomicInteger stateAfter = new AtomicInteger();
+
+ @Test
+ public void testBeforeFailsRunsAfterOfEarlierAdvices() throws Exception {
+ InternalProcessor ip = (InternalProcessor)
context.getRoute("start").getProcessor();
+ ip.addAdvice(new StatefulAdvice());
+ ip.addAdvice(new CamelInternalProcessorAdvice<Object>() {
+ @Override
+ public Object before(Exchange exchange) {
+ throw new IllegalStateException("Forced before");
+ }
+
+ @Override
+ public void after(Exchange exchange, Object data) {
+ throw new IllegalStateException("Should not run after when
before failed");
+ }
+
+ @Override
+ public boolean hasState() {
+ return false;
+ }
+ });
+
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ for (int i = 0; i < 3; i++) {
+ Exchange out = template.send("direct:start", e ->
e.getMessage().setBody("Hello"));
+ assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals("Forced before", out.getException().getMessage());
+ assertEquals(0, out.getException().getSuppressed().length);
+ }
+
+ assertMockEndpointsSatisfied();
+ // the exchanges must not be left inflight
+ assertEquals(0, context.getInflightRepository().size("start"));
+ assertEquals(0, context.getInflightRepository().size());
+ // the stateful advice gets its state
+ assertEquals(3, stateAfter.get());
+ }
+
+ @Test
+ public void testAfterFailsKeepsRouteException() {
+ InternalProcessor ip = (InternalProcessor)
context.getRoute("fail").getProcessor();
+ ip.addAdvice(new CamelInternalProcessorAdvice<Object>() {
+ @Override
+ public Object before(Exchange exchange) {
+ return null;
+ }
+
+ @Override
+ public void after(Exchange exchange, Object data) {
+ throw new IllegalStateException("Forced after");
+ }
+
+ @Override
+ public boolean hasState() {
+ return false;
+ }
+ });
+
+ Exchange out = template.send("direct:fail", e ->
e.getMessage().setBody("Hello"));
+ Exception cause = out.getException();
+ assertInstanceOf(IllegalArgumentException.class, cause);
+ assertEquals("Forced route", cause.getMessage());
+ assertEquals(1, cause.getSuppressed().length);
+ assertEquals("Forced after", cause.getSuppressed()[0].getMessage());
+ }
+
+ @Test
+ public void testAfterFailsWithoutRouteException() {
+ InternalProcessor ip = (InternalProcessor)
context.getRoute("start").getProcessor();
+ IllegalStateException forced = new IllegalStateException("Forced
after");
+ ip.addAdvice(new CamelInternalProcessorAdvice<Object>() {
+ @Override
+ public Object before(Exchange exchange) {
+ return null;
+ }
+
+ @Override
+ public void after(Exchange exchange, Object data) throws Exception
{
+ throw forced;
+ }
+
+ @Override
+ public boolean hasState() {
+ return false;
+ }
+ });
+
+ Exchange out = template.send("direct:start", e ->
e.getMessage().setBody("Hello"));
+ assertSame(forced, out.getException());
+ }
+
+ private class StatefulAdvice implements
CamelInternalProcessorAdvice<String> {
+ @Override
+ public String before(Exchange exchange) {
+ return "state";
+ }
+
+ @Override
+ public void after(Exchange exchange, String data) {
+ if ("state".equals(data)) {
+ stateAfter.incrementAndGet();
+ }
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("start")
+ .to("mock:result");
+
+ from("direct:fail").routeId("fail")
+ .throwException(new IllegalArgumentException("Forced
route"));
+ }
+ };
+ }
+}
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerAggregateStandbyTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerAggregateStandbyTest.java
new file mode 100644
index 000000000000..12c17ddbf01a
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerAggregateStandbyTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.builder.AggregationStrategies;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.debugger.BacklogTracer;
+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;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@DisabledOnOs(OS.AIX)
+public class BacklogTracerAggregateStandbyTest extends ManagementTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ // tracing is in standby (not enabled) and message history is not
enabled
+ context.setBacklogTracingStandby(true);
+ context.setMessageHistory(false);
+ return context;
+ }
+
+ @Test
+ public void testAggregateNotTracedInStandby() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("A,B,C");
+
+ template.sendBody("direct:start", "A");
+ template.sendBody("direct:start", "B");
+ template.sendBody("direct:start", "C");
+
+ assertMockEndpointsSatisfied();
+
+ BacklogTracer tracer =
context.getCamelContextExtension().getContextPlugin(BacklogTracer.class);
+ assertNotNull(tracer);
+ assertFalse(tracer.isEnabled());
+ assertEquals(0, tracer.getTraceCounter());
+ assertEquals(0, tracer.dumpAllTracedMessages().size());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("myRoute")
+
.aggregate(constant(true)).completionSize(3).aggregationStrategy(AggregationStrategies.string(","))
+ .id("aggregate")
+ .to("mock:result").id("result")
+ .end();
+ }
+ };
+ }
+
+}
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedProcessorSkipOverTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedProcessorSkipOverTest.java
new file mode 100644
index 000000000000..35820211f2a8
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedProcessorSkipOverTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.Exchange;
+import org.apache.camel.api.management.ManagedCamelContext;
+import org.apache.camel.api.management.mbean.ManagedProcessorMBean;
+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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@DisabledOnOs(OS.AIX)
+public class ManagedProcessorSkipOverTest extends ManagementTestSupport {
+
+ @Test
+ public void testSkipOverIsNotInflight() throws Exception {
+ getMockEndpoint("mock:skipped").expectedMessageCount(0);
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World",
"Hello World");
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ ManagedProcessorMBean mb =
context.getCamelContextExtension().getContextPlugin(ManagedCamelContext.class)
+ .getManagedProcessor("skipped");
+ assertNotNull(mb);
+ // the processor was skipped (such as step over in the debugger), so
it must not be left inflight
+ assertEquals(0L, mb.getExchangesInflight());
+ assertEquals(0L, mb.getExchangesFailed());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("foo")
+ // simulate the debugger skipping over the next
processor
+ .setProperty(Exchange.SKIP_OVER, constant(true))
+ .to("mock:skipped").id("skipped")
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 7efe7a4035ba..d2e5f405484b 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -64,6 +64,15 @@ The simple catalog (`simple.json`) now spells these
functions as they are writte
`exception.stacktrace` (was `exception.stackTrace`),
`throwException(msg,type)` (was `throwException(type,msg)`, the
order the function has always used), `not(exp)`, and `setAttachment(key,exp)`.
+=== camel-core - internal processor advices
+
+When a `CamelInternalProcessorAdvice` fails in its `before` method, then the
`after` method is now executed
+for the advices whose `before` was already executed. Previously they were
skipped, which could leave the exchange
+counted as inflight and its unit of work not done. The same now happens when
the debugger skips over a processor.
+
+An exception thrown from the `after` method of an advice no longer replaces
the exception the exchange has already
+failed with. Instead, it is added as a suppressed exception to the existing
exception.
+
=== Circuit Breaker EIP
The exchange property `CamelCircuitBreakerResponseRejected` is now also set
inside the `onFallback`,
