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 7acfd1c80a27 CAMEL-26888: camel-core - an Error of the aggregation
strategy of an enrich, or of a split watermark, fails the exchange (#29792)
7acfd1c80a27 is described below
commit 7acfd1c80a27550c4bcccae6a0856a28b01275d0
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 20:23:47 2026 +0530
CAMEL-26888: camel-core - an Error of the aggregation strategy of an
enrich, or of a split watermark, fails the exchange (#29792)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/camel/processor/Enricher.java | 8 +-
.../org/apache/camel/processor/PollEnricher.java | 8 +-
.../java/org/apache/camel/processor/Splitter.java | 8 +-
.../SplitterWatermarkThrowsErrorTest.java | 86 +++++++++++++++++++
...EnricherAggregationStrategyThrowsErrorTest.java | 98 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 +-
.../modules/ROOT/pages/error-handler.adoc | 14 +++-
7 files changed, 218 insertions(+), 11 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
index de329376788d..b4a9e53cf338 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java
@@ -254,8 +254,12 @@ public class Enricher extends BaseProcessorSupport
copyResultsWithoutCorrelationId(exchange,
aggregatedExchange);
}
used = true;
- } catch (Exception e) {
- // if the aggregationStrategy threw an exception, set
it on the original exchange
+ } catch (VirtualMachineError e) {
+ // left as before: the application must not handle a
JVM-level failure with Camel's error handling
+ throw e;
+ } catch (Throwable e) {
+ // if the aggregationStrategy threw an exception (also
an Error, which would escape this callback),
+ // set it on the original exchange
exchange.setException(new
CamelExchangeException("Error occurred during aggregation", exchange, e));
if (resourceExchange.getException() == null) {
resourceExchange.setException(e);
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
index 27f21ba74637..0df05c39cb16 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java
@@ -470,7 +470,11 @@ public class PollEnricher extends BaseProcessorSupport
implements IdAware, Route
// set property with the uri of the endpoint enriched so we can
use that for tracing etc
exchange.setProperty(ExchangePropertyKey.TO_ENDPOINT,
consumer.getEndpoint().getEndpointUri());
- } catch (Exception e) {
+ } catch (VirtualMachineError e) {
+ // left as before: the application must not handle a JVM-level
failure with Camel's error handling
+ throw e;
+ } catch (Throwable e) {
+ // also an Error of the aggregation strategy (such as a
NoClassDefFoundError), as for an exception
exchange.setException(new CamelExchangeException("Error occurred
during aggregation", exchange, e));
// the resource was not aggregated, so release it as failed now
(such as rolling back a polled file),
// and not together with the exchange, which may still complete
successfully (such as by redelivery)
@@ -489,7 +493,7 @@ public class PollEnricher extends BaseProcessorSupport
implements IdAware, Route
}
}
- private static void doneResourceAsFailed(Exchange resourceExchange,
Exception cause) {
+ private static void doneResourceAsFailed(Exchange resourceExchange,
Throwable cause) {
if (resourceExchange != null) {
if (cause != null && resourceExchange.getException() == null) {
resourceExchange.setException(cause);
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
index 0e2fb025bd47..fd8fd03733b7 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Splitter.java
@@ -737,8 +737,12 @@ public class Splitter extends MulticastProcessor {
String value;
try {
value = watermarkExpression.evaluate(subExchange,
String.class);
- } catch (Exception e) {
- // fail the item, as an exception thrown here would make
the split exchange hang (never complete)
+ } catch (VirtualMachineError e) {
+ // left as before: the application must not handle a
JVM-level failure with Camel's error handling
+ throw e;
+ } catch (Throwable e) {
+ // fail the item, as an exception (or an Error such as a
NoClassDefFoundError) thrown here would
+ // make the split exchange hang (never complete)
subExchange.setException(e);
return;
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkThrowsErrorTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkThrowsErrorTest.java
new file mode 100644
index 000000000000..9b2b9f68c441
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/SplitterWatermarkThrowsErrorTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.Arrays;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.SimpleFunction;
+import org.apache.camel.support.PluginHelper;
+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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * The watermark expression of a split throws an Error (not a
VirtualMachineError): the split completes and fails, as
+ * for an exception (CAMEL-26710).
+ */
+class SplitterWatermarkThrowsErrorTest extends ContextTestSupport {
+
+ private final Map<String, String> store = new ConcurrentHashMap<>();
+
+ @Test
+ void testSplitCompletesWhenTheWatermarkExpressionThrowsAnError() throws
Exception {
+ getMockEndpoint("mock:split").expectedMessageCount(2);
+
+ Exchange out = template.asyncSend("direct:start", e ->
e.getIn().setBody(Arrays.asList("a", "b")))
+ .get(5, TimeUnit.SECONDS);
+
+ MockEndpoint.assertIsSatisfied(context);
+ // the item failed with the Error (which the exchange keeps wrapped,
as for any Error)
+ assertNotNull(out.getException());
+ assertInstanceOf(NoClassDefFoundError.class,
out.getException().getCause());
+ assertFalse(store.containsKey("errorJob"));
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(0, context.getInflightRepository().size()));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a function of the watermark expression that needs a class
that is missing
+
PluginHelper.getSimpleFunctionRegistry(context).addFunction(new
SimpleFunction() {
+ @Override
+ public String getName() {
+ return "boom";
+ }
+
+ @Override
+ public Object apply(Exchange exchange, Object input) {
+ throw new NoClassDefFoundError("Simulated missing
class");
+ }
+ });
+
+ from("direct:start")
+ .split(body()).resumeStrategy(new
SplitterTestResumeStrategy(store), "errorJob")
+ .watermarkExpression("${function(boom)}")
+ .to("mock:split");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/enricher/EnricherAggregationStrategyThrowsErrorTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/enricher/EnricherAggregationStrategyThrowsErrorTest.java
new file mode 100644
index 000000000000..1337d366fdb7
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/enricher/EnricherAggregationStrategyThrowsErrorTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.enricher;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+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;
+
+/**
+ * The aggregation strategy of an enrich or a pollEnrich throws an Error (not
a VirtualMachineError): the exchange fails
+ * as for an exception of the strategy, and does not stay inflight.
+ */
+class EnricherAggregationStrategyThrowsErrorTest extends ContextTestSupport {
+
+ private final AggregationStrategy strategy = (oldExchange, newExchange) ->
{
+ throw new NoClassDefFoundError("Simulated missing class");
+ };
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ @Test
+ void testEnrich() throws Exception {
+
+ Exchange out = template.asyncSend("direct:enrich", e ->
e.getIn().setBody("Hello")).get(5, TimeUnit.SECONDS);
+
+ assertInstanceOf(CamelExchangeException.class, out.getException());
+ assertInstanceOf(NoClassDefFoundError.class,
out.getException().getCause());
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(0, context.getInflightRepository().size()));
+ }
+
+ @Test
+ void testPollEnrichReleasesTheFileAfterAnError() {
+ template.sendBodyAndHeader(fileUri("data"), "Big file",
Exchange.FILE_NAME, "AAA.dat");
+
+ // the first aggregation fails with an Error
+ Exchange out = template.send("direct:pollFile", e ->
e.getIn().setBody("Start"));
+ assertNotNull(out.getException());
+
+ // the file must be released, so the second poll gets it again
+ assertEquals("Big file", template.requestBody("direct:pollFile",
"Start", String.class));
+ }
+
+ @Test
+ void testPollEnrich() throws Exception {
+ template.sendBody("seda:resource", "World");
+
+ Exchange out = template.asyncSend("direct:pollEnrich", e ->
e.getIn().setBody("Hello")).get(5, TimeUnit.SECONDS);
+
+ assertInstanceOf(CamelExchangeException.class, out.getException());
+ assertInstanceOf(NoClassDefFoundError.class,
out.getException().getCause());
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(0, context.getInflightRepository().size()));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:enrich").enrich("direct:resource",
strategy).to("mock:result");
+ from("direct:resource").transform(constant("World"));
+
+
from("direct:pollFile").pollEnrich(fileUri("data?initialDelay=0&delay=10"),
2000, (original, resource) -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new NoClassDefFoundError("Simulated missing
class");
+ }
+
original.getMessage().setBody(resource.getMessage().getBody(String.class));
+ return original;
+ });
+
+ from("direct:pollEnrich").pollEnrich("seda:resource", 5000,
strategy).to("mock:result");
+ }
+ };
+ }
+}
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 c50fc5208253..6e3583d77b8d 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
@@ -896,7 +896,12 @@ exchange as an exception does (wrapped in a
`CamelExecutionException`), also whe
* by a transformer or validator of the input or output type of a route
(`inputType`, `outputType`). Before, it was
thrown to the caller of the route without the end of the unit of work, so
the exchange could stay inflight;
* by a plain `Processor` of a transacted exchange in a step without an error
handler of its own (such as in a `doTry`,
- `doCatch` or `doFinally` block). Before, a `doCatch` did not get it, and it
was thrown to the caller.
+ `doCatch` or `doFinally` block). Before, a `doCatch` did not get it, and it
was thrown to the caller;
+* by the `AggregationStrategy` of an `enrich` or `pollEnrich` (as a
`CamelExchangeException` caused by it, as for an
+ exception of the strategy), and the resource is released as failed. Before,
for `enrich` the `Error` was thrown to the
+ caller and the exchange stayed inflight; for `pollEnrich` the resource (such
as a polled file) was not released;
+* by the watermark expression of a `split`: it fails that item, as an
exception does. Before, it was thrown to the
+ caller and the split exchange stayed inflight.
A `VirtualMachineError` is left as before. See the next section.
diff --git a/docs/user-manual/modules/ROOT/pages/error-handler.adoc
b/docs/user-manual/modules/ROOT/pages/error-handler.adoc
index 99fb7f165360..63592b9d8e27 100644
--- a/docs/user-manual/modules/ROOT/pages/error-handler.adoc
+++ b/docs/user-manual/modules/ROOT/pages/error-handler.adoc
@@ -186,10 +186,12 @@ additional documentation of the features above.
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. The same rules
apply to an `Error` of the expression of
-a `recipientList`, of the expression of a `split` with `synchronous=true`, of
the `AggregationStrategy` of a `split`,
-`multicast` or `recipientList` with `synchronous=true` and without
`parallelProcessing`, of a data type transformer, of
-a transformer or validator of the input or output type of a route
(`inputType`, `outputType`), and of a plain
-`Processor` of a transacted exchange in a step without an error handler of its
own (such as in a `doTry` block):
+a `recipientList`, of the expression of a `split` with `synchronous=true`, of
the watermark expression of a `split`
+(it fails that item), of the `AggregationStrategy` of a `split`, `multicast`
or `recipientList` with
+`synchronous=true` and without `parallelProcessing`, of the
`AggregationStrategy` of an `enrich` or `pollEnrich`, of a
+data type transformer, of a transformer or validator of the input or output
type of a route (`inputType`,
+`outputType`), and of a plain `Processor` of a transacted exchange in a step
without an error handler of its own (such
+as in a `doTry` block):
[cols="2,4",options="header"]
|===
@@ -202,6 +204,10 @@ failure: redelivery, `onException`, `doCatch` and a dead
letter channel get it.
`doCatch(Exception.class)` also match it; to treat it differently, match it on
its class, such as
`onException(AssertionError.class)`
+For the `AggregationStrategy` of an `enrich` or `pollEnrich`, the exchange
fails with a `CamelExchangeException`
+caused by the `Error` (not wrapped in a `CamelExecutionException`), as for an
exception of the strategy, and the
+resource is released as failed
+
|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
