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 9a171984d3c0 CAMEL-26464: camel-core - an AsyncProcessor that throws
in a doTry or an onException fails the exchange instead of hanging (#29127)
9a171984d3c0 is described below
commit 9a171984d3c01a04c7e40f225e82b115e16e7dc8
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 4 14:11:32 2026 +0530
CAMEL-26464: camel-core - an AsyncProcessor that throws in a doTry or an
onException fails the exchange instead of hanging (#29127)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/engine/CamelInternalProcessor.java | 108 ++++-
.../apache/camel/impl/engine/DefaultChannel.java | 3 +
...thoutErrorHandlerProcessorThrowsPooledTest.java | 38 ++
...StepWithoutErrorHandlerProcessorThrowsTest.java | 226 ++++++++++
...houtErrorHandlerProcessorThrowsTestSupport.java | 453 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 15 +
6 files changed, 837 insertions(+), 6 deletions(-)
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 a43b8ce41b79..0fa6cd81b25a 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
@@ -23,6 +23,7 @@ import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
@@ -118,6 +119,9 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
private static final Object[] EMPTY_STATES = new Object[0];
+ private static final AtomicIntegerFieldUpdater<AsyncAfterTask> AWAITED
+ = AtomicIntegerFieldUpdater.newUpdater(AsyncAfterTask.class,
"awaited");
+
final CamelContext camelContext;
private final ReactiveExecutor reactiveExecutor;
private final ShutdownStrategy shutdownStrategy;
@@ -127,6 +131,10 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
// the id of the step, when this is the channel of a step that asks the
step hook itself, as it has no error handler
// that does (set by DefaultChannel)
String stepHookNodeId;
+ // whether the processor is the step itself, with no error handler around
it (set by DefaultChannel, such as for the
+ // steps of a doTry or an onException): this then catches an exception
that the processor throws instead of
+ // calling the callback, which would otherwise be dropped by the reactive
executor and leave the exchange hanging
+ boolean catchProcessorThrows;
public CamelInternalProcessor(CamelContext camelContext) {
this.camelContext = camelContext;
@@ -241,6 +249,12 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
private final Object[] states;
private Exchange exchange;
private AsyncCallback originalCallback;
+ // only when catchProcessorThrows: n while the callback of call n of
the processor with this task is still to
+ // come, and -n once call n is settled (0 before the first call). The
callback (done) and the takeover of a
+ // processor that throws before it calls the callback compete for it
(compareAndSet from n to -n), so the
+ // exchange is continued exactly once. The calls are numbered, as this
task is reused when tasks are pooled. One
+ // int field, not an object per call, as this task runs every step
(the field fits in the padding of the task)
+ volatile int awaited;
private AsyncAfterTask(Object[] states) {
this.states = states;
@@ -250,6 +264,16 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
public void prepare(Exchange exchange, AsyncCallback originalCallback)
{
this.exchange = exchange;
this.originalCallback = originalCallback;
+ if (catchProcessorThrows) {
+ int call = Math.abs(awaited) + 1;
+ if (call <= 0) {
+ // the numbers wrap around after Integer.MAX_VALUE calls
+ call = 1;
+ }
+ // an ordered write is enough: the callback reaches another
thread only through a hand-off that
+ // publishes it
+ AWAITED.lazySet(this, call);
+ }
}
@Override
@@ -266,6 +290,19 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
@Override
public void done(boolean doneSync) {
+ if (catchProcessorThrows) {
+ final int call = awaited;
+ if (call <= 0 || !AWAITED.compareAndSet(this, call, -call)) {
+ // the logs here do not read the exchange id: it is
created on first use, and the exchange may be
+ // in use on another thread
+ // (stepHookNodeId is the id of the step in a channel
without an error handler)
+ LOG.warn("The callback of {} (step {}) was called when it
was no longer awaited (after the processor"
+ + " had thrown an exception, which failed the
exchange instead, or more than once; an"
+ + " AsyncProcessor must not throw from process).
This callback is ignored.",
+ processor, stepHookNodeId);
+ return;
+ }
+ }
try {
AdviceIterator.runAfterTasks(advices, states, exchange);
} finally {
@@ -286,6 +323,44 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
}
}
}
+
+ /**
+ * Continues the exchange with the exception that the processor threw
before it called its callback, unless the
+ * callback came first.
+ */
+ void takeOver(int call, Throwable cause) {
+ if (!AWAITED.compareAndSet(this, call, -call)) {
+ // the callback continued the exchange (which may already be
done, or in use on another thread, so its
+ // id is not read here), so this exception is only logged
+ LOG.warn("Error processing an exchange by {} (step {}) due to
{}. The exchange continued, so this"
+ + " exception is ignored.",
+ processor, stepHookNodeId, cause.getMessage(), cause);
+ return;
+ }
+ // this task still holds the exchange and the callback of this
call: only the callback, by winning the
+ // compareAndSet, could have released it. Fail the exchange, as
when the processor sets the exception on
+ // the exchange and calls the callback
+ final AsyncCallback originalCallback = this.originalCallback;
+ exchange.setException(cause);
+ try {
+ AdviceIterator.runAfterTasks(advices, states, exchange);
+ } finally {
+ try {
+ // the after processing of the unit of work that the
callback of the processor would have done after
+ // this task (such as clearing the MDC)
+ final UnitOfWork uow = exchange.getUnitOfWork();
+ if (uow != null && uow.isBeforeAfterProcess()) {
+ uow.afterProcess(processor, exchange, this, false);
+ }
+ } finally {
+ reactiveExecutor.schedule(originalCallback);
+ }
+ // this task is not released: the processor may still call its
callback (this task), which must find no
+ // call awaited and be ignored, instead of continuing an
exchange that reuses this task. It keeps no
+ // reference to the exchange, which a broken processor may
hold on to with its callback
+ reset();
+ }
+ }
}
@Override
@@ -322,7 +397,7 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
Object[] states;
// create internal callback which will execute the advices in reverse
order when done
- CamelInternalTask afterTask = taskFactory != null ?
taskFactory.acquire() : null;
+ AsyncAfterTask afterTask = taskFactory != null ? (AsyncAfterTask)
taskFactory.acquire() : null;
if (afterTask == null) {
states = statefulAdvices > 0 ? new Object[statefulAdvices] :
EMPTY_STATES;
afterTask = new AsyncAfterTask(states);
@@ -381,7 +456,9 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
return true;
}
- private boolean processNonTransacted(Exchange exchange, CamelInternalTask
afterTask) {
+ private boolean processNonTransacted(Exchange exchange, AsyncAfterTask
afterTask) {
+ // read while this call owns the task: once the callback is called,
the task may be released and reused
+ final int call = catchProcessorThrows ? afterTask.awaited : 0;
final AsyncCallback async = beforeProcess(exchange, afterTask);
// ----------------------------------------------------------
@@ -391,10 +468,29 @@ public class CamelInternalProcessor extends
DelegateAsyncProcessor implements In
LOG.trace("Processing exchange for exchangeId: {} -> {}",
exchange.getExchangeId(), exchange);
}
final String nodeId = stepHookNodeId;
- boolean sync = nodeId == null
- ? processor.process(exchange, async)
- :
StepHelper.process(camelContext.getCamelContextExtension().getStepHook(),
exchange, nodeId, processor,
- async);
+ boolean sync;
+ try {
+ sync = nodeId == null
+ ? processor.process(exchange, async)
+ :
StepHelper.process(camelContext.getCamelContextExtension().getStepHook(),
exchange, nodeId, processor,
+ async);
+ } catch (Throwable e) {
+ // a Throwable, not only an Exception, as in the other fixes of
this kind (SimpleTask, RedeliveryTask,
+ // NoErrorHandler): an Error such as NoClassDefFoundError thrown
before the callback would also hang
+ if (!catchProcessorThrows || afterTask.awaited != call) {
+ // an error handler around the processor handles it, or the
callback was called before the exception was
+ // thrown and continued the exchange (this task may already be
released): leave the exception to the
+ // caller, as before
+ throw e;
+ }
+ // the processor threw before it called the callback, such as an
AsyncProcessor that throws from
+ // process(exchange, callback) instead of setting the exception on
the exchange. Take over through the
+ // reactive executor, after the work already queued there, so a
callback that the processor handed on and
+ // that this queued work calls still comes first and continues the
exchange (a callback that only work
+ // queued after the takeover calls loses to it, and is ignored)
+ reactiveExecutor.schedule(() -> afterTask.takeOver(call, e));
+ sync = false;
+ }
if (!sync) {
// Deprecated since 4.19.0
EventHelper.notifyExchangeAsyncProcessingStartedEvent(camelContext, exchange);
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
index efc46cb63f30..28f45d41c63a 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultChannel.java
@@ -131,6 +131,9 @@ public class DefaultChannel extends CamelInternalProcessor
implements Channel {
// the output has now been created, so assign the output as the
processor
setProcessor(getOutput());
+ // a step with no error handler (such as in a doTry or an onException)
is called directly, so this channel
+ // catches what its processor throws instead of calling the callback
+ catchProcessorThrows = errorHandler == null;
ServiceHelper.startService(errorHandler, output);
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsPooledTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsPooledTest.java
new file mode 100644
index 000000000000..83ab84acedd9
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsPooledTest.java
@@ -0,0 +1,38 @@
+/*
+ * 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 org.apache.camel.CamelContext;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.impl.engine.PooledProcessorExchangeFactory;
+
+/**
+ * {@link StepWithoutErrorHandlerProcessorThrowsTest} with pooled exchanges,
so the channels use pooled after tasks,
+ * which are reset and reused when they are released.
+ */
+class StepWithoutErrorHandlerProcessorThrowsPooledTest extends
StepWithoutErrorHandlerProcessorThrowsTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ExtendedCamelContext ecc = camelContext.getCamelContextExtension();
+ ecc.setExchangeFactory(new PooledExchangeFactory());
+ ecc.setProcessorExchangeFactory(new PooledProcessorExchangeFactory());
+ return camelContext;
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsTest.java
new file mode 100644
index 000000000000..130e4df1944d
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsTest.java
@@ -0,0 +1,226 @@
+/*
+ * 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.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Channel;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.support.AsyncProcessorSupport;
+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.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class StepWithoutErrorHandlerProcessorThrowsTest extends
StepWithoutErrorHandlerProcessorThrowsTestSupport {
+
+ @Test
+ void doTry() throws Exception {
+ assertSameAsSettingTheException(DO_TRY,
+ "exception=none caught=IllegalStateException:thrown result=0
caughtMock=1 finally=0 calls=1");
+ }
+
+ @Test
+ void doTryWithDoFinally() throws Exception {
+ assertSameAsSettingTheException(DO_TRY_FINALLY,
+ "exception=IllegalStateException:thrown
caught=IllegalStateException:thrown result=0 caughtMock=0 finally=1 calls=1");
+ }
+
+ @Test
+ void doTryWithDoFinallyProcessorThrowsError() throws Exception {
+ // an Error (such as NoClassDefFoundError) thrown before the callback
fails the exchange as well, as in the
+ // other fixes of this kind
+ start(DO_TRY_FINALLY, new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ throw new NoClassDefFoundError("thrown");
+ }
+ });
+ Exchange out = send();
+ assertInstanceOf(NoClassDefFoundError.class,
out.getException().getCause());
+ assertEquals("thrown", out.getException().getCause().getMessage());
+ assertEquals(1, calls.get(), "calls");
+ assertEquals(0,
getMockEndpoint("mock:start-result").getReceivedCounter(), "mock:result");
+ assertEquals(1,
getMockEndpoint("mock:start-finally").getReceivedCounter(), "mock:finally");
+ }
+
+ @Test
+ void doCatch() throws Exception {
+ assertSameAsSettingTheException(DO_CATCH,
+ "exception=IllegalStateException:thrown
caught=IllegalStateException:thrown result=0 caughtMock=0"
+ + " finally=1 calls=1");
+ }
+
+ @Test
+ void doFinally() throws Exception {
+ assertSameAsSettingTheException(DO_FINALLY,
+ "exception=IllegalStateException:thrown caught=none result=0
caughtMock=0 finally=0 calls=1");
+ }
+
+ @Test
+ void onException() throws Exception {
+ assertSameAsSettingTheException(ON_EXCEPTION,
+ "exception=IllegalStateException:thrown
caught=IllegalStateException:thrown result=0 caughtMock=0"
+ + " finally=0 calls=1");
+ }
+
+ @Test
+ void onExceptionNotHandled() throws Exception {
+ assertSameAsSettingTheException(ON_EXCEPTION_NOT_HANDLED,
+ "exception=IllegalStateException:thrown
caught=IllegalStateException:thrown result=0 caughtMock=0"
+ + "
finally=0 calls=1");
+ }
+
+ @Test
+ void multicastOutput() throws Exception {
+ // the failed output fails the exchange at the end of the multicast,
and the next output still runs
+ assertSameAsSettingTheException(MULTICAST,
+ "exception=IllegalStateException:thrown caught=none result=0
caughtMock=0 finally=1 calls=1");
+ }
+
+ @Test
+ void multicastOutputRedelivered() throws Exception {
+ // the output is redelivered as when it sets the exception, and
succeeds on the second call
+ assertSameAsSettingTheException(MULTICAST_REDELIVERY,
+ "exception=none caught=none result=1 caughtMock=0 finally=1
calls=2");
+ }
+
+ @Test
+ void afterAnAsynchronousStep() throws Exception {
+ // the step runs on the thread of the asynchronous delay, not on the
caller's
+ start((b, id, p) -> b
+ .from("direct:" + id)
+ .doTry()
+ .delay(10).asyncDelayed().process(p).to("mock:" + id +
"-result")
+ .endDoTry()
+ .doCatch(IllegalStateException.class)
+ .to("mock:" + id + "-caught")
+ .end(), new ThrowsFirst());
+
+ assertEquals("exception=none caught=IllegalStateException:thrown
result=0 caughtMock=1 finally=0 calls=1",
+ outcome(send()));
+ }
+
+ @Test
+ void withMdcLogging() throws Exception {
+ // with MDC logging the unit of work wraps the callback of each step
(the takeover runs its after processing
+ // instead of that wrapper)
+ context.setUseMDCLogging(true);
+ start(DO_TRY, new ThrowsFirst());
+
+ assertEquals("exception=none caught=IllegalStateException:thrown
result=0 caughtMock=1 finally=0 calls=1",
+ outcome(send()));
+ }
+
+ @Test
+ void controlTransacted() throws Exception {
+ // control: a transacted exchange calls the step synchronously, which
catches the exception, on main too
+ start(DO_TRY, new ThrowsFirst());
+
+ Exchange out = template.asyncSend("direct:start", e -> {
+ e.getMessage().setBody("Hello");
+ e.getExchangeExtension().setTransacted(true);
+ }).get(10, TimeUnit.SECONDS);
+ assertEquals("exception=none caught=IllegalStateException:thrown
result=0 caughtMock=1 finally=0 calls=1",
+ outcome(out));
+ }
+
+ @Test
+ void controlPlainProcessorThrows() throws Exception {
+ // control, as before: the converter of a plain Processor to an
AsyncProcessor catches the exception and sets it
+ // on the exchange
+ start(DO_TRY, exchange -> {
+ calls.incrementAndGet();
+ throw new IllegalStateException("thrown");
+ });
+
+ assertEquals("exception=none caught=IllegalStateException:thrown
result=0 caughtMock=1 finally=0 calls=1",
+ outcome(send()));
+ }
+
+ @Test
+ void controlCallbackCalledBeforeTheProcessorThrows() throws Exception {
+ // control, as before: the processor calls the callback and then
throws. The callback continued the exchange, so
+ // the exception is left to the caller (the reactive executor, which
logs it), and the exchange completes once
+ start(DO_TRY, new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ callback.done(true);
+ throw new IllegalStateException("after the callback");
+ }
+ });
+ AtomicInteger done = countDoneExchanges();
+
+ Exchange out = sendOnThisThread();
+ assertEquals("exception=none caught=none result=1 caughtMock=0
finally=0 calls=1", outcome(out));
+ assertEquals(1, done.get(), "the exchange was done once");
+ }
+
+ @Test
+ void
controlCallbackCalledBeforeTheProcessorThrowsReachesASynchronousCaller() throws
Exception {
+ // control, as before: a caller that calls the step through the
synchronous API (as a circuit breaker does with
+ // its steps) gets the exception that the processor throws after it
called the callback, as the callback only
+ // released that caller
+ Processor processor = new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ callback.done(true);
+ throw new IllegalStateException("after the callback");
+ }
+ };
+ start(DO_TRY, processor);
+ Channel channel = findChannel(context.getRoutes().get(0).navigate(),
processor);
+ assertNotNull(channel, "the channel of the step");
+
+ Exchange exchange =
context.getEndpoint("direct:start").createExchange();
+ IllegalStateException e = assertThrows(IllegalStateException.class, ()
-> channel.process(exchange));
+ assertEquals("after the callback", e.getMessage());
+ assertEquals(1, calls.get(), "calls");
+ }
+
+ @Test
+ void controlCallbackQueuedBeforeTheProcessorThrows() throws Exception {
+ // control, as before: the processor queues the call of the callback
on the reactive executor and then throws.
+ // The thrown failure is taken over through the reactive executor,
after the work already queued there, so the
+ // queued callback comes first and continues the exchange
+ start(DO_TRY, new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+
exchange.getContext().getCamelContextExtension().getReactiveExecutor().schedule(()
-> {
+ exchange.getMessage().setBody("queued");
+ callback.done(false);
+ });
+ throw new IllegalStateException("after the callback was
queued");
+ }
+ });
+ AtomicInteger done = countDoneExchanges();
+
+ Exchange out = sendOnThisThread();
+ assertEquals("exception=none caught=none result=1 caughtMock=0
finally=0 calls=1", outcome(out));
+ assertEquals("queued", out.getMessage().getBody());
+ assertEquals(1, done.get(), "the exchange was done once");
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsTestSupport.java
b/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsTestSupport.java
new file mode 100644
index 000000000000..7992db855d3f
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/StepWithoutErrorHandlerProcessorThrowsTestSupport.java
@@ -0,0 +1,453 @@
+/*
+ * 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.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Channel;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Navigate;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeCompletedEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeFailedEvent;
+import org.apache.camel.support.AsyncProcessorSupport;
+import org.apache.camel.support.EventNotifierSupport;
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+abstract class StepWithoutErrorHandlerProcessorThrowsTestSupport extends
ContextTestSupport {
+
+ final AtomicInteger calls = new AtomicInteger();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ /**
+ * Throws on the first call, and succeeds on the later calls.
+ */
+ final class ThrowsFirst extends AsyncProcessorSupport {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ if (calls.incrementAndGet() == 1) {
+ throw new IllegalStateException("thrown");
+ }
+ callback.done(true);
+ return true;
+ }
+ }
+
+ /**
+ * Routes from direct:{id}, to the mocks {id}-result, {id}-caught and
{id}-finally, so the processor that throws and
+ * the one that sets the exception can run side by side.
+ */
+ interface Routes {
+ void configure(RouteBuilder builder, String id, Processor processor);
+ }
+
+ static final Routes DO_TRY = (b, id, p) -> b
+ .from("direct:" + id)
+ .doTry()
+ .process(p).to("mock:" + id + "-result")
+ .doCatch(IllegalStateException.class)
+ .to("mock:" + id + "-caught")
+ .end();
+
+ static final Routes DO_TRY_FINALLY = (b, id, p) -> b
+ .from("direct:" + id)
+ .doTry()
+ .process(p).to("mock:" + id + "-result")
+ .doFinally()
+ .to("mock:" + id + "-finally")
+ .end();
+
+ static final Routes DO_CATCH = (b, id, p) -> b
+ .from("direct:" + id)
+ .doTry()
+ .throwException(new IllegalArgumentException("failed"))
+ .doCatch(IllegalArgumentException.class)
+ .process(p).to("mock:" + id + "-result")
+ .doFinally()
+ .to("mock:" + id + "-finally")
+ .end();
+
+ static final Routes DO_FINALLY = (b, id, p) -> b
+ .from("direct:" + id)
+ .doTry()
+ .to("mock:" + id + "-try")
+ .doFinally()
+ .process(p).to("mock:" + id + "-result")
+ .end();
+
+ static final Routes ON_EXCEPTION = (b, id, p) -> b
+ .from("direct:" + id)
+ .onException(IllegalArgumentException.class).handled(true)
+ .process(p).to("mock:" + id + "-result")
+ .end()
+ .throwException(new IllegalArgumentException("failed"));
+
+ static final Routes ON_EXCEPTION_NOT_HANDLED = (b, id, p) -> b
+ .from("direct:" + id)
+ .onException(IllegalArgumentException.class)
+ .process(p).to("mock:" + id + "-result")
+ .end()
+ .throwException(new IllegalArgumentException("failed"));
+
+ // the outputs of a multicast have no error handler in their channel (the
multicast wraps each one in a clone of
+ // the error handler of the route); the second output is the mock
{id}-finally, so the outcome shows whether it ran
+ static final Routes MULTICAST = (b, id, p) -> b
+ .from("direct:" + id)
+ .multicast()
+ .process(p)
+ .to("mock:" + id + "-finally")
+ .end()
+ .to("mock:" + id + "-result");
+
+ // the clone of the error handler around each output of a multicast
redelivers it
+ static final Routes MULTICAST_REDELIVERY = (b, id, p) -> b
+ .from("direct:" + id)
+
.errorHandler(b.defaultErrorHandler().maximumRedeliveries(2).redeliveryDelay(0))
+ .multicast()
+ .process(p)
+ .to("mock:" + id + "-finally")
+ .end()
+ .to("mock:" + id + "-result");
+
+ void start(Routes routes, Processor processor) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routes.configure(this, "start", processor);
+ }
+ });
+ context.start();
+ }
+
+ Exchange send() throws Exception {
+ return asyncSend().get(10, TimeUnit.SECONDS);
+ }
+
+ CompletableFuture<Exchange> asyncSend() {
+ // bounded by the caller: on a hang the future does not complete and
get times out
+ return template.asyncSend("direct:start", e ->
e.getMessage().setBody("Hello"));
+ }
+
+ /**
+ * Sends on this thread, which runs the reactive work of the exchange
until none is left: when it returns, the
+ * takeover of a thrown failure has run too.
+ */
+ Exchange sendOnThisThread() {
+ return template.send("direct:start", e ->
e.getMessage().setBody("Hello"));
+ }
+
+ /**
+ * Counts the exchanges done: a second continuation of an exchange would
complete its unit of work again.
+ */
+ AtomicInteger countDoneExchanges() {
+ AtomicInteger done = new AtomicInteger();
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ done.incrementAndGet();
+ }
+
+ @Override
+ public boolean isEnabled(CamelEvent event) {
+ return event instanceof ExchangeCompletedEvent || event
instanceof ExchangeFailedEvent;
+ }
+ });
+ return done;
+ }
+
+ /**
+ * What a route does with an exchange: the failure, the caught exception
and the mocks reached.
+ */
+ String outcome(Exchange out) {
+ return outcome(out, "start", calls);
+ }
+
+ String outcome(Exchange out, String id, AtomicInteger calls) {
+ Exception caught = out.getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+ return "exception=" + describe(out.getException()) + " caught=" +
describe(caught)
+ + " result=" + getMockEndpoint("mock:" + id +
"-result").getReceivedCounter()
+ + " caughtMock=" + getMockEndpoint("mock:" + id +
"-caught").getReceivedCounter()
+ + " finally=" + getMockEndpoint("mock:" + id +
"-finally").getReceivedCounter()
+ + " calls=" + calls.get();
+ }
+
+ static String describe(Throwable t) {
+ return t == null ? "none" : t.getClass().getSimpleName() + ":" +
t.getMessage();
+ }
+
+ /**
+ * The route with the processor that sets the exception on the exchange
(from direct:sets) and with the one that
+ * throws (from direct:start) must do the same.
+ */
+ void assertSameAsSettingTheException(Routes routes, String expected)
throws Exception {
+ AtomicInteger setsCalls = new AtomicInteger();
+ Processor sets = new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ if (setsCalls.incrementAndGet() == 1) {
+ exchange.setException(new IllegalStateException("thrown"));
+ }
+ callback.done(true);
+ return true;
+ }
+ };
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routes.configure(this, "sets", sets);
+ routes.configure(this, "start", new ThrowsFirst());
+ }
+ });
+ context.start();
+
+ // the exchanges done (with the copies of a multicast): a second
continuation would complete one again
+ AtomicInteger done = countDoneExchanges();
+ Exchange reference = template.asyncSend("direct:sets", e ->
e.getMessage().setBody("Hello")).get(10,
+ TimeUnit.SECONDS);
+ assertEquals(expected, outcome(reference, "sets", setsCalls), "the
processor that sets the exception");
+ await().atMost(10, TimeUnit.SECONDS).until(() -> done.get() >= 1);
+ int referenceDone = done.getAndSet(0);
+
+ Exchange out = send();
+ assertEquals(expected, outcome(out), "the processor that throws");
+ await().atMost(10, TimeUnit.SECONDS).until(() -> done.get() >=
referenceDone);
+ assertEquals(referenceDone, done.get(), "the exchanges done, as with
the processor that sets the exception");
+ }
+
+ @Test
+ void nextExchangeIsProcessed() throws Exception {
+ start(DO_TRY, new ThrowsFirst());
+ assertEquals("exception=none caught=IllegalStateException:thrown
result=0 caughtMock=1 finally=0 calls=1",
+ outcome(send()));
+
+ Exchange out = send();
+ assertNull(out.getException());
+ assertNull(out.getProperty(Exchange.EXCEPTION_CAUGHT));
+ assertEquals(1,
getMockEndpoint("mock:start-result").getReceivedCounter(), "mock:result");
+ assertEquals(1,
getMockEndpoint("mock:start-caught").getReceivedCounter(), "mock:caught");
+ assertEquals(2, calls.get(), "calls");
+ }
+
+ static Channel findChannel(Navigate<Processor> nav, Processor processor) {
+ if (nav == null || !nav.hasNext()) {
+ return null;
+ }
+ for (Processor p : nav.next()) {
+ if (p instanceof Channel channel && channel.getNextProcessor() ==
processor) {
+ return channel;
+ }
+ if (p instanceof Navigate<?>) {
+ @SuppressWarnings("unchecked")
+ Channel found = findChannel((Navigate<Processor>) p,
processor);
+ if (found != null) {
+ return found;
+ }
+ }
+ }
+ return null;
+ }
+
+ @Test
+ void callbackCalledAfterTheProcessorThrewIsIgnored() throws Exception {
+ // the processor hands the callback to another thread and then throws.
The step cannot tell this from a callback
+ // that never comes, so it fails the exchange with the thrown
exception (before, the exchange hung), and the
+ // late callback must not continue the exchange a second time
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicReference<Thread> late = new AtomicReference<>();
+ AtomicReference<Throwable> lateFailure = new AtomicReference<>();
+ start(DO_TRY, new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ Thread thread = new Thread(() -> {
+ try {
+ if (release.await(10, TimeUnit.SECONDS)) {
+ callback.done(false);
+ }
+ } catch (Throwable t) {
+ lateFailure.set(t);
+ }
+ }, "late-callback");
+ late.set(thread);
+ thread.start();
+ throw new IllegalStateException("thrown");
+ }
+ });
+ AtomicInteger done = countDoneExchanges();
+
+ assertEquals("exception=none caught=IllegalStateException:thrown
result=0 caughtMock=1 finally=0 calls=1",
+ outcome(send()));
+
+ // now the processor calls the callback, after the exchange completed
+ release.countDown();
+ late.get().join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(late.get().isAlive(), "the late callback thread ended");
+ assertNull(lateFailure.get(), "the late callback failed");
+ assertEquals(1, done.get(), "the exchange was done once");
+ assertEquals(0,
getMockEndpoint("mock:start-result").getReceivedCounter(), "mock:result");
+ assertEquals(1,
getMockEndpoint("mock:start-caught").getReceivedCounter(), "mock:caught");
+ }
+
+ @Test
+ void callbackCalledAfterTheProcessorThrewDoesNotContinueTheNextExchange()
throws Exception {
+ // as above, but the late callback comes while the next exchange waits
for the callback of its own call of the
+ // processor: it must not continue that exchange
+ CountDownLatch releaseFirst = new CountDownLatch(1);
+ CountDownLatch releaseSecond = new CountDownLatch(1);
+ AtomicReference<Thread> first = new AtomicReference<>();
+ AtomicReference<Thread> second = new AtomicReference<>();
+ AtomicReference<Throwable> lateFailure = new AtomicReference<>();
+ start(DO_TRY, new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ int call = calls.incrementAndGet();
+ CountDownLatch release = call == 1 ? releaseFirst :
releaseSecond;
+ Thread thread = new Thread(() -> {
+ try {
+ if (release.await(10, TimeUnit.SECONDS)) {
+ callback.done(false);
+ }
+ } catch (Throwable t) {
+ lateFailure.set(t);
+ }
+ }, "callback-" + call);
+ (call == 1 ? first : second).set(thread);
+ thread.start();
+ if (call == 1) {
+ throw new IllegalStateException("thrown");
+ }
+ return false;
+ }
+ });
+
+ assertEquals("exception=none caught=IllegalStateException:thrown
result=0 caughtMock=1 finally=0 calls=1",
+ outcome(send()));
+
+ CompletableFuture<Exchange> next = asyncSend();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> second.get() != null);
+
+ // the callback of the first call, while the second call waits for its
own
+ releaseFirst.countDown();
+ first.get().join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(first.get().isAlive(), "the late callback thread ended");
+ assertFalse(next.isDone(), "the late callback of the first exchange
continued the next exchange");
+ assertEquals(0,
getMockEndpoint("mock:start-result").getReceivedCounter(), "mock:result");
+
+ releaseSecond.countDown();
+ Exchange out = next.get(10, TimeUnit.SECONDS);
+ assertNull(out.getException());
+ assertEquals(1,
getMockEndpoint("mock:start-result").getReceivedCounter(), "mock:result");
+ second.get().join(TimeUnit.SECONDS.toMillis(10));
+ assertNull(lateFailure.get(), "a callback failed");
+ assertEquals(2, calls.get(), "calls");
+ }
+
+ @Test
+ void callbackAndThrowAtTheSameTimeContinueEachExchangeOnce() throws
Exception {
+ // the processor hands the callback to another thread, which calls it
at once, and throws: the callback and the
+ // takeover race, and whichever wins, each exchange is continued once
(as a success, or caught by the doCatch)
+ int count = 50;
+ ExecutorService callbacks = Executors.newFixedThreadPool(2);
+ try {
+ start(DO_TRY, new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback
callback) {
+ calls.incrementAndGet();
+ callbacks.execute(() -> callback.done(false));
+ throw new IllegalStateException("thrown");
+ }
+ });
+ AtomicInteger done = countDoneExchanges();
+
+ List<CompletableFuture<Exchange>> sends = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ sends.add(asyncSend());
+ }
+ int succeeded = 0;
+ int caught = 0;
+ for (CompletableFuture<Exchange> send : sends) {
+ Exchange out = send.get(20, TimeUnit.SECONDS);
+ assertNull(out.getException());
+ Exception e = out.getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+ if (e == null) {
+ succeeded++;
+ } else {
+ assertEquals("thrown", e.getMessage());
+ caught++;
+ }
+ }
+ await().atMost(10, TimeUnit.SECONDS).until(() -> done.get() >=
count);
+ // the callbacks that lost have run too
+ callbacks.shutdown();
+ assertTrue(callbacks.awaitTermination(10, TimeUnit.SECONDS), "the
callbacks ran");
+
+ assertEquals(count, done.get(), "each exchange was done once");
+ assertEquals(succeeded,
getMockEndpoint("mock:start-result").getReceivedCounter(), "mock:result");
+ assertEquals(caught,
getMockEndpoint("mock:start-caught").getReceivedCounter(), "mock:caught");
+ assertEquals(count, calls.get(), "calls");
+ } finally {
+ callbacks.shutdownNow();
+ }
+ }
+
+ @Test
+ void callbackCalledTwiceContinuesTheExchangeOnce() throws Exception {
+ // a processor that calls its callback twice (also a broken
processor): the second call is ignored
+ AtomicReference<Throwable> secondCall = new AtomicReference<>();
+ start(DO_TRY, new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ callback.done(true);
+ try {
+ callback.done(true);
+ } catch (Throwable t) {
+ secondCall.set(t);
+ }
+ return true;
+ }
+ });
+ AtomicInteger done = countDoneExchanges();
+
+ Exchange out = sendOnThisThread();
+ assertEquals("exception=none caught=none result=1 caughtMock=0
finally=0 calls=1", outcome(out));
+ assertNull(secondCall.get(), "the second call of the callback failed");
+ assertEquals(1, done.get(), "the exchange was done once");
+ }
+}
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 6ce3498cfcdb..435bf93c39fe 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
@@ -434,6 +434,21 @@ caller), put the EIP in a `doTry` with only a `doFinally`
in the called route, o
called route with `errorHandler(noErrorHandler())`, which turns it off for the
other steps of that route too. Neither
option brings back the redelivery of a whole routing slip or dynamic router.
+=== camel-core - an AsyncProcessor that throws in a step without an error
handler fails the exchange
+
+A step that has no error handler of its own (the steps of `doTry`, `doCatch`,
`doFinally` and `onException`, also when
+nested in another EIP inside them, the outputs of a multicast, the steps of a
circuit breaker, and a step with
+`inheritErrorHandler(false)`) now handles a processor that throws from
`process(exchange, callback)` instead of
+setting the exception on the exchange as if it had set it: the exchange fails
with that exception, so the `doCatch`
+catches it, and the output of a multicast is redelivered by the error handler
of the route as for any other failure.
+Before, the exchange hung.
+
+When such a processor handed its callback to another thread and then throws,
the exchange now fails with the thrown
+exception unless that thread calls the callback first, and a later call is
ignored and logged at WARN; before, the
+exception was dropped and that callback continued the exchange as a success. A
second call of the callback is ignored
+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 - 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,
