This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26741-camel-bean-vme in repository https://gitbox.apache.org/repos/asf/camel.git
commit 4432d7cc6de8a38b0f0e8c5bdceb6ece709c834f Author: Claus Ibsen <[email protected]> AuthorDate: Tue Nov 10 21:25:38 2026 +0100 CAMEL-26741: camel-bean - a VirtualMachineError thrown by a bean method is handled as for a Processor Follow-up to #29525, which leaves a VirtualMachineError (OutOfMemoryError, StackOverflowError, InternalError) of a plain Processor as before: it is not set on the exchange, but thrown on after the callback, so Camel's error handling does not get a JVM-level failure. A plain bean method did the opposite: reflection wraps any Throwable in an InvocationTargetException, and MethodInfo set its target on the exchange, so onException, doCatch, redelivery and the dead letter channel also handled an OutOfMemoryError. MethodInfo now throws a VirtualMachineError on after calling the callback, as the Processor adapters do. Any other Error still fails the exchange. An Error thrown outside the reflective call (such as from a Callable the bean method returned) is now also set on the exchange, instead of escaping before the callback. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../apache/camel/component/bean/MethodInfo.java | 22 +- .../component/bean/BeanMethodThrowsErrorTest.java | 228 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 16 ++ 3 files changed, 260 insertions(+), 6 deletions(-) diff --git a/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfo.java b/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfo.java index 0a6f2c7673be..7cea9f6dc860 100644 --- a/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfo.java +++ b/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfo.java @@ -270,14 +270,24 @@ public class MethodInfo { MessageHelper.resetStreamCache(exchange.getIn()); return doProceed(callback); } catch (InvocationTargetException e) { - exchange.setException(e.getTargetException()); - callback.done(true); - return true; - } catch (Exception e) { - exchange.setException(e); + // the bean method threw: reflection wraps any Throwable, a VirtualMachineError included + return failed(e.getTargetException(), callback); + } catch (Throwable e) { + // any Throwable, not only an Exception, such as an Error from a returned Callable + return failed(e, callback); + } + } + + private boolean failed(Throwable e, AsyncCallback callback) { + if (e instanceof VirtualMachineError vme) { + // as a Processor in the bean component (and DelegateSyncProcessor): a JVM-level failure such as an + // OutOfMemoryError is not given to Camel's error handling, but thrown on after the callback callback.done(true); - return true; + throw vme; } + exchange.setException(e); + callback.done(true); + return true; } private boolean doProceed(AsyncCallback callback) throws Exception { diff --git a/core/camel-core/src/test/java/org/apache/camel/component/bean/BeanMethodThrowsErrorTest.java b/core/camel-core/src/test/java/org/apache/camel/component/bean/BeanMethodThrowsErrorTest.java new file mode 100644 index 000000000000..14245f2d8f7e --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/bean/BeanMethodThrowsErrorTest.java @@ -0,0 +1,228 @@ +/* + * 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.bean; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelExecutionException; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.support.service.ServiceHelper; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A bean method that throws an {@link Error} is handled as a bean that is a {@link org.apache.camel.Processor}: an + * {@link Error} such as an {@link AssertionError} fails the exchange and the error handler gets it, while a + * {@link VirtualMachineError} (such as an {@link OutOfMemoryError} or a {@link StackOverflowError}) is not given to + * Camel's error handling (onException, doCatch, the dead letter channel, redelivery) but thrown on after the callback. + */ +public class BeanMethodThrowsErrorTest extends ContextTestSupport { + + private static final ProcessingError ERROR = new ProcessingError(); + // an OutOfMemoryError that reaches JUnit aborts the whole test run, so it is only thrown where assertThrows expects it + private static final OutOfMemoryError OOM = new OutOfMemoryError("test"); + private static final StackOverflowError SOE = new StackOverflowError("test"); + + private final AtomicInteger calls = new AtomicInteger(); + + @Test + public void beanMethodThatThrowsAnErrorFailsTheExchange() throws Exception { + getMockEndpoint("mock:result").expectedMessageCount(0); + + Exchange out = template.send("direct:error", e -> e.getMessage().setBody("Hello")); + + assertFailedWithTheError(out.getException()); + assertMockEndpointsSatisfied(); + assertEquals(0, context.getInflightRepository().size(), "The exchange is still inflight"); + } + + @Test + public void errorHandlingAppliesToAnErrorOfABeanMethod() throws Exception { + getMockEndpoint("mock:handled").expectedBodiesReceived("Hello"); + getMockEndpoint("mock:result").expectedMessageCount(0); + + Exchange out = template.send("direct:handled", e -> e.getMessage().setBody("Hello")); + + assertFailedWithTheError(out.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class)); + assertMockEndpointsSatisfied(); + } + + @Test + public void errorOfACallableReturnedByABeanMethodFailsTheExchange() throws Exception { + getMockEndpoint("mock:result").expectedMessageCount(0); + + Exchange out = template.send("direct:callable", e -> e.getMessage().setBody("Hello")); + + assertFailedWithTheError(out.getException()); + assertMockEndpointsSatisfied(); + assertEquals(0, context.getInflightRepository().size(), "The exchange is still inflight"); + } + + @Test + public void stackOverflowErrorOfABeanMethodIsNotGivenToOnException() throws Exception { + getMockEndpoint("mock:handled").expectedMessageCount(0); + + Exchange out = template.send("direct:vm-handled", e -> e.getMessage().setBody("Hello")); + + // as a bean that is a Processor: the error handler logs the Error and continues the exchange, without onException + assertNull(out.getException()); + assertEquals(1, calls.get()); + assertMockEndpointsSatisfied(); + } + + @Test + public void outOfMemoryErrorOfABeanMethodIsNotGivenToDoCatch() throws Exception { + // as a bean that is a Processor, the exchange stays inflight, so do not wait for it when the context stops + context.getShutdownStrategy().setTimeUnit(TimeUnit.MILLISECONDS); + context.getShutdownStrategy().setTimeout(100); + getMockEndpoint("mock:caught").expectedMessageCount(0); + + OutOfMemoryError thrown = assertThrows(OutOfMemoryError.class, + () -> template.send("direct:vm-try", e -> e.getMessage().setBody("Hello"))); + + assertSame(OOM, thrown); + assertMockEndpointsSatisfied(); + } + + @Test + public void stackOverflowErrorOfABeanMethodIsNotRedeliveredNorGivenToTheDeadLetterChannel() throws Exception { + getMockEndpoint("mock:dead").expectedMessageCount(0); + + Exchange out = template.send("direct:vm-dlc", e -> e.getMessage().setBody("Hello")); + + assertNull(out.getException()); + assertEquals(1, calls.get(), "The exchange was redelivered"); + assertMockEndpointsSatisfied(); + } + + @Test + public void beanProcessorCallsTheCallbackOnceWithTheError() throws Exception { + BeanProcessor processor = startedBeanProcessor(new ThrowingBean(ERROR, calls)); + Exchange exchange = createExchangeWithBody("Hello"); + List<Boolean> done = new ArrayList<>(); + + boolean sync = processor.process(exchange, done::add); + + assertTrue(sync); + assertEquals(List.of(true), done); + assertFailedWithTheError(exchange.getException()); + } + + @Test + public void beanProcessorThrowsAVirtualMachineErrorOnAfterTheCallback() throws Exception { + BeanProcessor processor = startedBeanProcessor(new ThrowingBean(OOM, calls)); + Exchange exchange = createExchangeWithBody("Hello"); + List<Boolean> done = new ArrayList<>(); + + OutOfMemoryError thrown = assertThrows(OutOfMemoryError.class, () -> processor.process(exchange, done::add)); + + assertSame(OOM, thrown); + assertEquals(List.of(true), done); + assertNull(exchange.getException()); + } + + private BeanProcessor startedBeanProcessor(Object bean) { + BeanProcessor processor = new BeanProcessor(bean, context); + processor.setMethod("doSomething"); + ServiceHelper.startService(processor); + return processor; + } + + private static void assertFailedWithTheError(Exception exception) { + // the exchange keeps an Error wrapped in a CamelExecutionException + assertIsInstanceOf(CamelExecutionException.class, exception); + assertSame(ERROR, exception.getCause()); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:error") + .bean(new ThrowingBean(ERROR, calls), "doSomething") + .to("mock:result"); + + from("direct:handled") + .onException(ProcessingError.class).handled(true).to("mock:handled").end() + .bean(new ThrowingBean(ERROR, calls), "doSomething") + .to("mock:result"); + + from("direct:callable") + .bean(new ThrowingBean(ERROR, calls), "callable") + .to("mock:result"); + + from("direct:vm-handled") + .onException(Throwable.class).handled(true).to("mock:handled").end() + .bean(new ThrowingBean(SOE, calls), "doSomething"); + + from("direct:vm-try") + .doTry() + .bean(new ThrowingBean(OOM, calls), "doSomething") + .doCatch(Throwable.class) + .to("mock:caught") + .end(); + + from("direct:vm-dlc") + .errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(3).redeliveryDelay(0)) + .bean(new ThrowingBean(SOE, calls), "doSomething"); + } + }; + } + + public static final class ThrowingBean { + private final Error error; + private final AtomicInteger calls; + + ThrowingBean(Error error, AtomicInteger calls) { + this.error = error; + this.calls = calls; + } + + public void doSomething(String body) { + calls.incrementAndGet(); + throw error; + } + + public Callable<String> callable(String body) { + // the Callable is called by the bean component after the method returned, outside the reflective call + return () -> { + calls.incrementAndGet(); + throw error; + }; + } + } + + private static final class ProcessingError extends Error { + private static final long serialVersionUID = 1L; + + ProcessingError() { + super("Simulated error of a bean method"); + } + } +} 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 b74611bbcb8a..fcb047c9dc20 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 @@ -3251,6 +3251,22 @@ The `aws2-ddbstream` consumer now reads a closed shard to its end before it read shard rotation the consumer started again from `streamIteratorType`: with `FROM_START` it routed records of the stream again, and with `FROM_LATEST` it skipped records already written to the new shards. +=== camel-bean - a VirtualMachineError thrown by a bean method is no longer given to the error handler + +A bean method that throws a `VirtualMachineError` (such as `OutOfMemoryError`, `StackOverflowError` or +`InternalError`) is now handled as a bean that is a `Processor`: Camel's error handling does not get it (the JVM may no +longer be in a consistent state, so the application should not handle it). It is not set on the exchange, but thrown on +after the callback of the bean was called. Depending on the error handler, the `Error` reaches the caller (and the +exchange may stay inflight), or it is only logged and the route continues as if the bean method had succeeded. + +Before, the bean component set any `Error` of a bean method on the exchange, so `onException`, `doCatch`, redelivery +and a dead letter channel also handled an `OutOfMemoryError` or a `StackOverflowError`. Any other `Error` of a bean +method, such as an `AssertionError`, still fails the exchange and goes to the error handler, as before. + +An application with strict reliability needs should not rely on Camel to react to a `VirtualMachineError`. For an +`OutOfMemoryError` that the JVM raises, it can let the JVM stop with `-XX:+ExitOnOutOfMemoryError` or +`-XX:+CrashOnOutOfMemoryError`. + === camel-beanio - BeanIOSplitter decodes files and streams with the charset of the exchange or its encoding `BeanIOSplitter` read a file body with the default charset of the JVM, and a stream or byte array body with the
