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 a909f0224a98 CAMEL-26741: camel-core - an Error thrown by a plain
Processor fails the exchange (#29525)
a909f0224a98 is described below
commit a909f0224a98023fd48c47bd97aa3e43ba5f85b0
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 11 12:20:27 2026 +0530
CAMEL-26741: camel-core - an Error thrown by a plain Processor fails the
exchange (#29525)
* CAMEL-26741: camel-core - an Error thrown by a plain Processor fails the
exchange
* CAMEL-26741: upgrade guide - a test that asserts inside a processor and
sends with template.send must check the returned exchange
* CAMEL-26741: camel-core - leave a VirtualMachineError as before, and
document which Errors fail the exchange
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/sanitizer.adoc | 12 +-
.../component/bean/AbstractBeanProcessor.java | 7 +-
.../camel-sanitizer/src/main/docs/sanitizer.adoc | 12 +-
.../java/org/apache/camel/sanitizer/Sanitizer.java | 52 +---
.../camel/sanitizer/SanitizerStepListener.java | 13 -
.../camel/processor/ProcessorThrowsErrorTest.java | 327 +++++++++++++++++++++
.../org/apache/camel/sanitizer/BrokenUserCode.java | 2 +-
.../camel/sanitizer/SanitizerUserCodeTest.java | 71 ++---
.../support/AsyncProcessorConverterHelper.java | 8 +-
.../support/processor/DelegateSyncProcessor.java | 9 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 43 +++
.../modules/ROOT/pages/error-handler.adoc | 33 +++
12 files changed, 463 insertions(+), 126 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sanitizer.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sanitizer.adoc
index 591e85cc05cd..555cac11156c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sanitizer.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sanitizer.adoc
@@ -48,12 +48,12 @@ fails the exchange with what was thrown, and the callback
and that failure race
An exception that a processor throws on through an outer step is reported
once, at the step that threw it first.
-An `Error` (such as an `AssertionError` of a test) that a plain `Processor`
throws breaks the rule in Camel, not in the
-processor: the adapter Camel puts around a plain `Processor`
(`DelegateSyncProcessor` for `process(...)`, the bridge of
-`AsyncProcessorConverterHelper`, and the bean component for a bean that is a
`Processor`) calls the callback in
-`finally` and then lets the `Error` through. It is reported as
`THREW_AFTER_CALLBACK` at that step, naming the
-processor, and the report says the callback was called by Camel's adapter and
that the fix is in what threw the
-`Error`. Without stacks (`stackDepth=0`) the adapter cannot be seen, and the
report is the one of an `AsyncProcessor`.
+A plain `Processor` does not break the rule when it throws: the adapter Camel
puts around it (`DelegateSyncProcessor`
+for `process(...)`, the bridge of `AsyncProcessorConverterHelper`, and the
bean component for a bean that is a
+`Processor`) sets what it throws on the exchange, also an `Error` such as an
`AssertionError` of a test, and then calls
+the callback. The exchange fails, and nothing is reported. A
`VirtualMachineError` (such as an `OutOfMemoryError` or
+a `StackOverflowError`) is the exception: the adapter does not catch it, so it
is thrown on after the callback, as in
+earlier versions, and is reported as thrown after the callback
(`THREW_AFTER_CALLBACK`).
A report looks like this:
diff --git
a/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java
b/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java
index 5b909e7a495f..4ab28bef031c 100644
---
a/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java
+++
b/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java
@@ -158,7 +158,12 @@ public abstract class AbstractBeanProcessor extends
AsyncProcessorSupport {
}
try {
target.process(exchange);
- } catch (AssertionError | Exception e) {
+ } catch (VirtualMachineError e) {
+ // left as before (thrown on after the callback): the application
must not handle a JVM-level failure such
+ // as an OutOfMemoryError with Camel's error handling
+ throw e;
+ } catch (Throwable e) {
+ // any other Throwable, as in DelegateSyncProcessor: the callback
is called in finally
exchange.setException(e);
} finally {
callback.done(true);
diff --git a/components/camel-sanitizer/src/main/docs/sanitizer.adoc
b/components/camel-sanitizer/src/main/docs/sanitizer.adoc
index 591e85cc05cd..555cac11156c 100644
--- a/components/camel-sanitizer/src/main/docs/sanitizer.adoc
+++ b/components/camel-sanitizer/src/main/docs/sanitizer.adoc
@@ -48,12 +48,12 @@ fails the exchange with what was thrown, and the callback
and that failure race
An exception that a processor throws on through an outer step is reported
once, at the step that threw it first.
-An `Error` (such as an `AssertionError` of a test) that a plain `Processor`
throws breaks the rule in Camel, not in the
-processor: the adapter Camel puts around a plain `Processor`
(`DelegateSyncProcessor` for `process(...)`, the bridge of
-`AsyncProcessorConverterHelper`, and the bean component for a bean that is a
`Processor`) calls the callback in
-`finally` and then lets the `Error` through. It is reported as
`THREW_AFTER_CALLBACK` at that step, naming the
-processor, and the report says the callback was called by Camel's adapter and
that the fix is in what threw the
-`Error`. Without stacks (`stackDepth=0`) the adapter cannot be seen, and the
report is the one of an `AsyncProcessor`.
+A plain `Processor` does not break the rule when it throws: the adapter Camel
puts around it (`DelegateSyncProcessor`
+for `process(...)`, the bridge of `AsyncProcessorConverterHelper`, and the
bean component for a bean that is a
+`Processor`) sets what it throws on the exchange, also an `Error` such as an
`AssertionError` of a test, and then calls
+the callback. The exchange fails, and nothing is reported. A
`VirtualMachineError` (such as an `OutOfMemoryError` or
+a `StackOverflowError`) is the exception: the adapter does not catch it, so it
is thrown on after the callback, as in
+earlier versions, and is reported as thrown after the callback
(`THREW_AFTER_CALLBACK`).
A report looks like this:
diff --git
a/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/Sanitizer.java
b/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/Sanitizer.java
index b1048e3a5e75..35a3e9a1ad41 100644
---
a/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/Sanitizer.java
+++
b/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/Sanitizer.java
@@ -23,7 +23,6 @@ import java.util.EnumMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledExecutorService;
@@ -56,7 +55,6 @@ import org.apache.camel.model.SplitDefinition;
import org.apache.camel.model.ToDefinition;
import org.apache.camel.model.ToDynamicDefinition;
import org.apache.camel.spi.IdAware;
-import org.apache.camel.support.AsyncProcessorConverterHelper;
import org.apache.camel.support.processor.DelegateAsyncProcessor;
import org.apache.camel.support.processor.DelegateSyncProcessor;
import org.apache.camel.util.URISupport;
@@ -248,20 +246,12 @@ public final class Sanitizer {
}
/**
- * Reports a violation: counts it, logs it at WARN and charges it to the
test session.
+ * Reports a violation: counts it, logs it at WARN and charges it to the
test session. The first 1000 are kept for
+ * the report and logged; the others are counted and still charged to the
test session, so a test that breaks the
+ * rule fails also when a long-lived context (such as a cached test
context) passed the bound before it.
*/
void record(
Violation violation, String exchangeId, String routeId, String
stepId, String details) {
- record(violation, exchangeId, routeId, stepId, details,
violation.getFix());
- }
-
- /**
- * Reports a violation with its own usual fix. The first 1000 are kept for
the report and logged; the others are
- * counted and still charged to the test session, so a test that breaks
the rule fails also when a long-lived
- * context (such as a cached test context) passed the bound before it.
- */
- void record(
- Violation violation, String exchangeId, String routeId, String
stepId, String details, String fix) {
counts.get(violation).increment();
boolean keep = kept.incrementAndGet() <= MAX_KEPT;
if (!keep && overflowLogged.compareAndSet(false, true)) {
@@ -277,7 +267,7 @@ public final class Sanitizer {
@Override
public SanitizerViolation get() {
if (built == null) {
- built = violationOf(violation, exchangeId, routeId,
stepId, details, fix);
+ built = violationOf(violation, exchangeId, routeId,
stepId, details);
}
return built;
}
@@ -294,7 +284,7 @@ public final class Sanitizer {
}
private SanitizerViolation violationOf(
- Violation violation, String exchangeId, String routeId, String
stepId, String details, String fix) {
+ Violation violation, String exchangeId, String routeId, String
stepId, String details) {
String processor = processorOf(routeId, stepId);
StringBuilder sb = new StringBuilder(512);
sb.append("Exchange sanitizer:
").append(violation.getTitle()).append(" (R1)\n");
@@ -302,7 +292,7 @@ public final class Sanitizer {
sb.append(details);
sb.append(" The callback of process(exchange, callback) must be
called exactly once, and process must not throw"
+ " (see AsyncProcessor).\n");
- sb.append(" Usual fix: ").append(fix).append('.');
+ sb.append(" Usual fix: ").append(violation.getFix()).append('.');
return new SanitizerViolation(
violation, camelContext.getName(), exchangeId, routeId,
stepId, processor, sb.toString());
}
@@ -815,17 +805,6 @@ public final class Sanitizer {
*/
static final class Trace {
- /**
- * The methods of Camel that call a plain {@link Processor} and then
its callback in {@code finally}, catching
- * only exceptions: an {@link Error} of the processor goes through
after the callback (by name, as the bridge is
- * a private nested class, binary name with {@code $}, and the bean
one is in camel-bean; a rename must be
- * followed here, which SanitizerUserCodeTest checks).
- */
- static final Set<String> SYNC_ADAPTERS = Set.of(
- DelegateSyncProcessor.class.getName() + ".process",
- AsyncProcessorConverterHelper.class.getName() +
"$ProcessorToAsyncProcessorBridge.process",
-
"org.apache.camel.component.bean.AbstractBeanProcessor.useCustomAdapter");
-
private final String thread;
private final StackTraceElement[] frames;
@@ -834,25 +813,6 @@ public final class Sanitizer {
this.frames = frames;
}
- /**
- * Whether the callback was called by an adapter Camel puts around a
plain {@link Processor}, and not by the
- * code of the processor: the first frame that is not a callback (a
{@code done} method or a lambda of one) is
- * one of {@link #SYNC_ADAPTERS}. Unknown (false) without stacks
({@code stackDepth=0}).
- */
- boolean calledBySyncAdapter() {
- if (frames == null) {
- return false;
- }
- for (StackTraceElement f : frames) {
- String method = f.getMethodName();
- if ("done".equals(method) || method.startsWith("lambda$")) {
- continue;
- }
- return SYNC_ADAPTERS.contains(f.getClassName() + "." + method);
- }
- return false;
- }
-
void appendTo(StringBuilder sb, String what) {
sb.append(" ").append(what).append(": thread
").append(thread).append('\n');
if (frames != null) {
diff --git
a/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/SanitizerStepListener.java
b/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/SanitizerStepListener.java
index 904cd39977f3..34a37ead8dc2 100644
---
a/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/SanitizerStepListener.java
+++
b/components/camel-sanitizer/src/main/java/org/apache/camel/sanitizer/SanitizerStepListener.java
@@ -81,12 +81,6 @@ final class SanitizerStepListener implements StepListener {
*/
static final class StepCall {
- // the usual fix when a plain Processor threw an Error after Camel's
adapter called the callback
- static final String SYNC_ADAPTER_FIX
- = "fix what threw the Error. The processor is a plain
Processor, so the callback is not its own: Camel's"
- + " adapter around it (such as DelegateSyncProcessor for
process(...)) calls the callback in finally and"
- + " then lets the Error through";
-
private final Sanitizer sanitizer;
private final String exchangeId;
private final String routeId;
@@ -187,13 +181,6 @@ final class SanitizerStepListener implements StepListener {
if (first == null) {
trace.appendTo(sb, "process threw " + description);
sanitizer.record(Violation.THREW_BEFORE_CALLBACK, exchangeId,
routeId, nodeId, sb.toString());
- } else if (cause instanceof Error && first.calledBySyncAdapter()) {
- // a plain Processor threw an Error: Camel's adapter called
the callback in finally and let the Error
- // through, so the adapter breaks the rule, not the processor
- first.appendTo(sb, "the callback was called by Camel's adapter
for a plain Processor");
- trace.appendTo(sb, "then the Error of the processor went
through the adapter: " + description);
- sanitizer.record(Violation.THREW_AFTER_CALLBACK, exchangeId,
routeId, nodeId, sb.toString(),
- SYNC_ADAPTER_FIX);
} else {
first.appendTo(sb, "the callback was called");
trace.appendTo(sb, "then process threw " + description);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/ProcessorThrowsErrorTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/ProcessorThrowsErrorTest.java
new file mode 100644
index 000000000000..99848ed31fab
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/ProcessorThrowsErrorTest.java
@@ -0,0 +1,327 @@
+/*
+ * 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.processor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.support.AsyncProcessorConverterHelper;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.processor.DelegateSyncProcessor;
+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 plain {@link Processor} may throw an {@link Error}, such as an {@link
AssertionError} of a test. The adapters that
+ * call it as an {@link AsyncProcessor} must then fail the exchange with it
and call the callback once, as for an
+ * exception, instead of calling the callback and throwing the error on. A
{@link VirtualMachineError} (such as an
+ * {@link OutOfMemoryError} or a {@link StackOverflowError}) is left as
before: the adapters do not set it on the
+ * exchange, so Camel's error handling (onException, doCatch, the dead letter
channel, redelivery) does not get it.
+ */
+public class ProcessorThrowsErrorTest 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 processorThatThrowsAnErrorFailsTheExchange() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ Exchange out = template.send("direct:process", e ->
e.getMessage().setBody("Hello"));
+
+ assertFailedWithTheError(out.getException());
+ assertMockEndpointsSatisfied();
+ assertEquals(0, context.getInflightRepository().size(), "The exchange
is still inflight");
+ }
+
+ @Test
+ public void errorHandlingAppliesToAnErrorOfAProcessor() 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();
+ assertEquals(0, context.getInflightRepository().size(), "The exchange
is still inflight");
+ }
+
+ @Test
+ public void doCatchCatchesAnErrorOfAProcessor() throws Exception {
+ getMockEndpoint("mock:caught").expectedBodiesReceived("Hello");
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ Exchange out = template.send("direct:try", e ->
e.getMessage().setBody("Hello"));
+
+ assertNull(out.getException());
+ assertMockEndpointsSatisfied();
+ assertEquals(0, context.getInflightRepository().size(), "The exchange
is still inflight");
+ }
+
+ @Test
+ public void beanThatIsAProcessorAndThrowsAnErrorFailsTheExchange() throws
Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ Exchange out = template.send("direct:bean", e ->
e.getMessage().setBody("Hello"));
+
+ assertFailedWithTheError(out.getException());
+ assertMockEndpointsSatisfied();
+ assertEquals(0, context.getInflightRepository().size(), "The exchange
is still inflight");
+ }
+
+ @Test
+ public void delegateSyncProcessorCallsTheCallbackOnceWithTheError() {
+ assertCallbackCalledOnceWithTheError(new DelegateSyncProcessor(new
ThrowingProcessor()));
+ }
+
+ @Test
+ public void convertedProcessorCallsTheCallbackOnceWithTheError() {
+
assertCallbackCalledOnceWithTheError(AsyncProcessorConverterHelper.convert(new
ThrowingProcessor()));
+ }
+
+ @Test
+ public void stackOverflowErrorOfAProcessorIsNotGivenToOnException() throws
Exception {
+ getMockEndpoint("mock:handled").expectedMessageCount(0);
+
+ Exchange out = template.send("direct:vm-process", e ->
e.getMessage().setBody("Hello"));
+
+ // as before: the error handler logs the Error and continues the
exchange, without onException
+ assertNull(out.getException());
+ assertEquals(1, calls.get());
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void outOfMemoryErrorOfAProcessorIsNotGivenToDoCatch() throws
Exception {
+ // as before, the exchange stays inflight, so do not wait for it when
the context stops
+ shortShutdownTimeout();
+ 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
stackOverflowErrorOfAProcessorIsNotRedeliveredNorGivenToTheDeadLetterChannel()
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
stackOverflowErrorOfABeanThatIsAProcessorIsNotGivenToOnException() throws
Exception {
+ getMockEndpoint("mock:handled").expectedMessageCount(0);
+
+ Exchange out = template.send("direct:vm-bean", e ->
e.getMessage().setBody("Hello"));
+
+ assertNull(out.getException());
+ assertEquals(1, calls.get());
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void stackOverflowErrorOfAProducerIsNotGivenToOnException() throws
Exception {
+ getMockEndpoint("mock:handled").expectedMessageCount(0);
+
+ Exchange out = template.send("direct:vm-producer", e ->
e.getMessage().setBody("Hello"));
+
+ assertNull(out.getException());
+ assertEquals(1, calls.get());
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void adaptersThrowAVirtualMachineErrorOnAfterTheCallback() {
+ assertVirtualMachineErrorThrownOnAfterTheCallback(new
DelegateSyncProcessor(new ThrowingProcessor(OOM, calls)));
+ assertVirtualMachineErrorThrownOnAfterTheCallback(
+ AsyncProcessorConverterHelper.convert(new
ThrowingProcessor(OOM, calls)));
+ }
+
+ private void
assertVirtualMachineErrorThrownOnAfterTheCallback(AsyncProcessor processor) {
+ 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 void shortShutdownTimeout() {
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.MILLISECONDS);
+ context.getShutdownStrategy().setTimeout(100);
+ }
+
+ private void assertCallbackCalledOnceWithTheError(AsyncProcessor
processor) {
+ Exchange exchange = createExchangeWithBody("Hello");
+ List<Boolean> calls = new ArrayList<>();
+
+ boolean sync = processor.process(exchange, calls::add);
+
+ assertTrue(sync);
+ assertEquals(List.of(true), calls);
+ assertFailedWithTheError(exchange.getException());
+ }
+
+ 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() throws Exception {
+ from("direct:process")
+ .process(new ThrowingProcessor())
+ .to("mock:result");
+
+ from("direct:handled")
+
.onException(ProcessingError.class).handled(true).to("mock:handled").end()
+ .process(new ThrowingProcessor())
+ .to("mock:result");
+
+ from("direct:try")
+ .doTry()
+ .process(new ThrowingProcessor())
+ .to("mock:result")
+ .doCatch(ProcessingError.class)
+ .to("mock:caught")
+ .end();
+
+ from("direct:bean")
+ .bean(new ThrowingProcessor())
+ .to("mock:result");
+
+ from("direct:vm-process")
+
.onException(Throwable.class).handled(true).to("mock:handled").end()
+ .process(new ThrowingProcessor(SOE, calls));
+
+ from("direct:vm-try")
+ .doTry()
+ .process(new ThrowingProcessor(OOM, calls))
+ .doCatch(Throwable.class)
+ .to("mock:caught")
+ .end();
+
+ from("direct:vm-dlc")
+
.errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(3).redeliveryDelay(0))
+ .process(new ThrowingProcessor(SOE, calls));
+
+ from("direct:vm-bean")
+
.onException(Throwable.class).handled(true).to("mock:handled").end()
+ .bean(new ThrowingProcessor(SOE, calls));
+
+ getContext().addEndpoint("throwing:producer", new
ThrowingProducerEndpoint(calls));
+ from("direct:vm-producer")
+
.onException(Throwable.class).handled(true).to("mock:handled").end()
+ .to("throwing:producer");
+ }
+ };
+ }
+
+ private static final class ThrowingProcessor implements Processor {
+ private final Error error;
+ private final AtomicInteger calls;
+
+ ThrowingProcessor() {
+ this(ERROR, new AtomicInteger());
+ }
+
+ ThrowingProcessor(Error error, AtomicInteger calls) {
+ this.error = error;
+ this.calls = calls;
+ }
+
+ @Override
+ public void process(Exchange exchange) {
+ calls.incrementAndGet();
+ throw error;
+ }
+ }
+
+ /**
+ * An endpoint whose producer is not an AsyncProcessor, so it is called
through the bridge of
+ * AsyncProcessorConverterHelper.
+ */
+ private static final class ThrowingProducerEndpoint extends
DefaultEndpoint {
+ private final AtomicInteger calls;
+
+ ThrowingProducerEndpoint(AtomicInteger calls) {
+ this.calls = calls;
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new DefaultProducer(this) {
+ @Override
+ public void process(Exchange exchange) {
+ calls.incrementAndGet();
+ throw SOE;
+ }
+ };
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException("Producer only");
+ }
+
+ @Override
+ protected String createEndpointUri() {
+ return "throwing:producer";
+ }
+ }
+
+ private static final class ProcessingError extends Error {
+ private static final long serialVersionUID = 1L;
+
+ ProcessingError() {
+ super("Simulated error of a processor");
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/sanitizer/BrokenUserCode.java
b/core/camel-core/src/test/java/org/apache/camel/sanitizer/BrokenUserCode.java
index de36b657be9b..e8a80976313c 100644
---
a/core/camel-core/src/test/java/org/apache/camel/sanitizer/BrokenUserCode.java
+++
b/core/camel-core/src/test/java/org/apache/camel/sanitizer/BrokenUserCode.java
@@ -193,7 +193,7 @@ final class BrokenUserCode {
/**
* A plain Processor (not an AsyncProcessor) that throws an Error, as an
assertion of a test in a processor does.
- * Camel's adapter around it calls the callback, and lets the Error
through.
+ * Camel's adapter around it sets the Error on the exchange and calls the
callback, so it breaks no rule.
*/
static final class ThrowsError implements Processor {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/sanitizer/SanitizerUserCodeTest.java
b/core/camel-core/src/test/java/org/apache/camel/sanitizer/SanitizerUserCodeTest.java
index 8ed6de2ba2f7..714b5e1f0658 100644
---
a/core/camel-core/src/test/java/org/apache/camel/sanitizer/SanitizerUserCodeTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/sanitizer/SanitizerUserCodeTest.java
@@ -17,11 +17,11 @@
package org.apache.camel.sanitizer;
import java.time.Duration;
-import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.camel.CamelContext;
+import org.apache.camel.CamelExecutionException;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
@@ -31,12 +31,15 @@ import org.junit.jupiter.api.Test;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Rule R1 on processors and strategies of a user that break the contract of
AsyncProcessor: each test fails with a
- * report that names the step and the class, and says so with {@link
ExpectSanitizerViolation}.
+ * report that names the step and the class, and says so with {@link
ExpectSanitizerViolation}. The tests without it
+ * check what Camel handles so that no rule is broken, such as an Error (not a
VirtualMachineError) of a plain
+ * Processor.
*/
@CamelSanitizer
class SanitizerUserCodeTest extends ContextTestSupport {
@@ -65,7 +68,7 @@ class SanitizerUserCodeTest extends ContextTestSupport {
// no id: the step has the id Camel generates, as process1
from("direct:autoId").routeId("autoId").process(new
BrokenUserCode.ThrowsInProcess());
from("direct:plainError").process(new
BrokenUserCode.ThrowsError()).id("plainError");
- from("direct:beanError").bean(new
BrokenUserCode.ThrowsError()).id("beanError");
+ from("direct:beanError").routeId("beanRoute").bean(new
BrokenUserCode.ThrowsError()).id("beanError");
from("direct:late").process(throwsThenCallsBack).id("late");
from("direct:aggregateError")
.aggregate(constant(true), new
BrokenUserCode.ErrorStrategy()).completionSize(2).id("aggregate")
@@ -160,57 +163,27 @@ class SanitizerUserCodeTest extends ContextTestSupport {
}
@Test
- @ExpectSanitizerViolation(value = Violation.THREW_AFTER_CALLBACK, step =
"plainError")
- void anErrorOfAPlainProcessorIsReportedAsCamelsAdapter() {
- sendAndCatchTheError("direct:plainError");
-
- SanitizerViolation v = awaitOne(Violation.THREW_AFTER_CALLBACK);
- assertEquals(BrokenUserCode.ThrowsError.class.getName(),
v.getProcessor());
- assertAdapterReport(v);
- }
-
- @Test
- @ExpectSanitizerViolation(value = Violation.THREW_AFTER_CALLBACK, step =
"beanError")
- void aBeanIsNamedByItsClass() {
- sendAndCatchTheError("direct:beanError");
-
- SanitizerViolation v = awaitOne(Violation.THREW_AFTER_CALLBACK);
- // the bean, not Camel's BeanProcessor that calls it
- assertEquals(BrokenUserCode.ThrowsError.class.getName(),
v.getProcessor());
- assertAdapterReport(v);
+ void anErrorOfAPlainProcessorFailsTheExchange() {
+ // Camel's adapter around a plain Processor sets what it throws on the
exchange, also an Error, and then calls the
+ // callback: no rule is broken
+ assertFailedWithTheError(template.send("direct:plainError", e ->
e.getMessage().setBody("x")));
+ assertEquals(0, Sanitizer.get(context).getViolations().size());
}
@Test
- void theAdaptersOfCamelAreFoundByTheirNames() throws Exception {
- // the adapters are known by class and method name (one is a private
nested class, one is in camel-bean): a
- // rename or a move of one must be followed in
Sanitizer.Trace.SYNC_ADAPTERS
- for (String adapter : Sanitizer.Trace.SYNC_ADAPTERS) {
- int dot = adapter.lastIndexOf('.');
- Class<?> type = Class.forName(adapter.substring(0, dot));
- String method = adapter.substring(dot + 1);
- assertTrue(Arrays.stream(type.getDeclaredMethods()).anyMatch(m ->
m.getName().equals(method)), adapter);
- }
- }
-
- private void sendAndCatchTheError(String uri) {
- // the exchange stays inflight after the Error went through, so the
stop of the context does not wait the
- // default 10 seconds for it
- context.getShutdownStrategy().setTimeout(1);
- try {
- template.send(uri, e -> e.getMessage().setBody("x"));
- } catch (BrokenUserCode.StrategyBug e) {
- // the Error goes through Camel's adapter, after the callback;
only the report matters here
- }
+ void anErrorOfABeanThatIsAProcessorFailsTheExchange() {
+ assertFailedWithTheError(template.send("direct:beanError", e ->
e.getMessage().setBody("x")));
+ assertEquals(0, Sanitizer.get(context).getViolations().size());
+ // a report at this step would name the bean, not Camel's
BeanProcessor that calls it
+ assertEquals(BrokenUserCode.ThrowsError.class.getName(),
+ Sanitizer.get(context).processorOf("beanRoute", "beanError"));
}
- private static void assertAdapterReport(SanitizerViolation v) {
- // the callback was called by Camel's adapter around the plain
Processor, not by the processor: the report says
- // so, and does not tell the author of the processor to stop throwing
after the callback
- assertTrue(v.getText().contains("the callback was called by Camel's
adapter for a plain Processor"), v.getText());
-
assertTrue(v.getText().contains(BrokenUserCode.StrategyBug.class.getName() + ":
Forced by ThrowsError"),
- v.getText());
- assertTrue(v.getText().contains("Usual fix: " +
SanitizerStepListener.StepCall.SYNC_ADAPTER_FIX), v.getText());
- assertTrue(v.getText().contains("at " +
BrokenUserCode.ThrowsError.class.getName() + ".process"), v.getText());
+ private static void assertFailedWithTheError(Exchange out) {
+ // the exchange keeps the Error wrapped in a CamelExecutionException
+ assertInstanceOf(CamelExecutionException.class, out.getException());
+ assertInstanceOf(BrokenUserCode.StrategyBug.class,
out.getException().getCause());
+ assertEquals("Forced by ThrowsError",
out.getException().getCause().getMessage());
}
@Test
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/AsyncProcessorConverterHelper.java
b/core/camel-support/src/main/java/org/apache/camel/support/AsyncProcessorConverterHelper.java
index 455f582f3f65..4dd62586250b 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/AsyncProcessorConverterHelper.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/AsyncProcessorConverterHelper.java
@@ -63,8 +63,12 @@ public final class AsyncProcessorConverterHelper {
}
try {
processor.process(exchange);
- } catch (Exception e) {
- // must catch throwable so we catch all
+ } catch (VirtualMachineError e) {
+ // left as before (thrown on after the callback): the
application must not handle a JVM-level failure
+ // such as an OutOfMemoryError with Camel's error handling
+ throw e;
+ } catch (Throwable e) {
+ // any other Throwable, as in DelegateSyncProcessor: the
callback is called in finally
exchange.setException(e);
} finally {
// we are bridging a sync processor as async so callback with
true
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/processor/DelegateSyncProcessor.java
b/core/camel-support/src/main/java/org/apache/camel/support/processor/DelegateSyncProcessor.java
index 92c2e1979f87..60341233bb97 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/processor/DelegateSyncProcessor.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/processor/DelegateSyncProcessor.java
@@ -65,8 +65,13 @@ public class DelegateSyncProcessor extends ServiceSupport
// force calling the sync method
try {
processor.process(exchange);
- } catch (Exception | LinkageError e) {
- // we catch all exceptions and try to catch relatively manageable
low-level errors
+ } catch (VirtualMachineError e) {
+ // left as before (thrown on after the callback): the application
must not handle a JVM-level failure such
+ // as an OutOfMemoryError with Camel's error handling
+ throw e;
+ } catch (Throwable e) {
+ // any other Throwable, also an Error such as an AssertionError:
the callback is called in finally, so an
+ // Error thrown on after it would not fail the exchange, and the
exchange could stay inflight
exchange.setException(e);
} finally {
// we are bridging a sync processor as async so callback with true
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 fcb047c9dc20..bb0dc8cbdd84 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
@@ -649,6 +649,49 @@ exception was dropped and that callback continued the
exchange as a success. A s
and logged at WARN; before, the exchange was continued a second time. There is
no option for the old behaviour; fix the
processor to set the exception on the exchange and call the callback, as an
`AsyncProcessor` must.
+=== camel-core - an Error thrown by a Processor fails the exchange
+
+When a plain `Processor` (in `process(...)`, in `bean(...)` for a bean that is
a `Processor`, or as a producer that is
+not an `AsyncProcessor`) throws an `Error` that is not a
`VirtualMachineError`, such as an `AssertionError`, the
+exchange now fails with it, as for an exception. Before, only a `LinkageError`
thrown in `process(...)` and an
+`AssertionError` thrown by a bean that is a `Processor` did so (as a bean
method that throws any `Error` already does).
+Any other `Error` was thrown on after the processor's callback had been
called: depending on the error handler, it was
+only logged and the next steps ran as if the processor had succeeded, or it
reached the caller and the exchange stayed
+inflight until the context stopped (which then waited for its shutdown
timeout).
+
+[cols="2,4",options="header"]
+|===
+|Thrown by the processor |What happens
+
+|An `Error` that is not a `VirtualMachineError`, such as `AssertionError`,
`LinkageError`
+(`NoClassDefFoundError`, `ExceptionInInitializerError`) or a custom `Error`
+|The exchange fails with it (wrapped in a `CamelExecutionException`), and the
error handler handles it as any other
+failure: redelivery, `onException`, `doCatch` and a dead letter channel get
it. `onException(Exception.class)` and
+`doCatch(Exception.class)` also match it; to treat it differently, match it on
its class, such as
+`onException(AssertionError.class)`
+
+|A `VirtualMachineError`, such as `OutOfMemoryError`, `StackOverflowError` or
`InternalError`
+|Not changed by this entry: 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 processor's
+callback 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 processor
had succeeded
+|===
+
+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` (these options do not cover a
`StackOverflowError`, or an `OutOfMemoryError` that code
+throws itself).
+
+`template.send(...)`, `template.request(...)` and `template.asyncSend(...)`
return the exchange (or its future) with the
+exception and do not throw it. `sendBody(...)` and `requestBody(...)` throw a
`CamelExecutionException` that holds the
+`Error`, unless the failure was handled (such as by a dead letter channel,
`onException(...).handled(true)` or
+`doCatch`). A test that sends with `template.send(...)` and asserts only
inside a processor
+must now check the exception of the returned exchange
(`exchange.getException()`), or send with `sendBody(...)` or
+`requestBody(...)`.
+
+There is no option for the old behaviour. See also
+xref:error-handler.adoc#_an_error_thrown_by_a_processor[An Error thrown by a
processor].
+
=== camel-core - variableReceive into a route or group variable stores the
header variables in that route or group
With `variableReceive` (`toV`, `enrich`, `pollEnrich`, `fromV` and so on) into
a route variable or a group variable,
diff --git a/docs/user-manual/modules/ROOT/pages/error-handler.adoc
b/docs/user-manual/modules/ROOT/pages/error-handler.adoc
index bdb39e90e4f7..b35127b3006e 100644
--- a/docs/user-manual/modules/ROOT/pages/error-handler.adoc
+++ b/docs/user-manual/modules/ROOT/pages/error-handler.adoc
@@ -182,6 +182,39 @@ Dead Letter Channel
See xref:exception-clause.adoc[Exception Clause] documentation for
additional documentation of the features above.
+== An Error thrown by a processor
+
+A plain `Processor` (in `process(...)`, in `bean(...)` for a bean that is a
`Processor`, or a producer that is not an
+`AsyncProcessor`) may throw an `Error` instead of an exception:
+
+[cols="2,4",options="header"]
+|===
+|Thrown by the processor |What happens
+
+|An `Error` that is not a `VirtualMachineError`, such as `AssertionError`,
`LinkageError`
+(`NoClassDefFoundError`, `ExceptionInInitializerError`) or a custom `Error`
+|The exchange fails with it (wrapped in a `CamelExecutionException`), and the
error handler handles it as any other
+failure: redelivery, `onException`, `doCatch` and a dead letter channel get
it. `onException(Exception.class)` and
+`doCatch(Exception.class)` also match it; to treat it differently, match it on
its class, such as
+`onException(AssertionError.class)`
+
+|A `VirtualMachineError`, such as `OutOfMemoryError`, `StackOverflowError` or
`InternalError`
+|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 processor's
+callback 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 processor
had succeeded
+|===
+
+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` (these options do not cover a
`StackOverflowError`, or an `OutOfMemoryError` that code
+throws itself).
+
+`template.send(...)`, `template.request(...)` and `template.asyncSend(...)`
return the failed exchange (or its future)
+without throwing the exception; `sendBody(...)` and `requestBody(...)` throw a
`CamelExecutionException` that holds the
+`Error`, unless the failure was handled (such as by a dead letter channel,
`onException(...).handled(true)` or
+`doCatch`).
+
== Scopes
The error handler is scoped as either:
