This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24978 in repository https://gitbox.apache.org/repos/asf/camel.git
commit e17a1ba372a9f2b86d7494821c273b573ce0ac40 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Sep 23 22:16:22 2026 +0200 CAMEL-24978: camel-core - Internal processor runs after advices on skip over and failed before The debugger skip over and an advice failing in before, did not run the after of the advices whose before was executed, leaving exchanges inflight (route and JMX) and message history and tracer state unfinished. An exception thrown from an after advice no longer replaces the exception of the exchange, but is added as suppressed. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/impl/engine/AdviceIterator.java | 29 +++- .../camel/impl/engine/CamelInternalProcessor.java | 20 ++- .../engine/CamelInternalProcessorAdviceTest.java | 154 +++++++++++++++++++++ .../management/ManagedProcessorSkipOverTest.java | 64 +++++++++ 4 files changed, 259 insertions(+), 8 deletions(-) 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..8c5a14a13eb1 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 @@ -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 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/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"); + } + }; + } +}
