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 5551dee680b2 CAMEL-26446: camel-core - RedeliveryErrorHandler: a throw
on the first delivery goes through redelivery, the dead letter channel and
onException, and continues the exchange once (#29040)
5551dee680b2 is described below
commit 5551dee680b2136f728980456f8abb23856b740d
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 3 14:17:14 2026 +0530
CAMEL-26446: camel-core - RedeliveryErrorHandler: a throw on the first
delivery goes through redelivery, the dead letter channel and onException, and
continues the exchange once (#29040)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../errorhandler/RedeliveryErrorHandler.java | 112 +++++-
...veryFirstDeliveryProcessorThrowsPooledTest.java | 38 ++
...RedeliveryFirstDeliveryProcessorThrowsTest.java | 407 +++++++++++++++++++++
...eryFirstDeliveryProcessorThrowsTestSupport.java | 78 ++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 19 +
5 files changed, 640 insertions(+), 14 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
index 9f7f7792b095..5c05406e6a86 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
@@ -1173,12 +1173,23 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
* Task to perform calling the processor and handling redelivery if it
fails (more advanced than ProcessTask)
*/
protected class RedeliveryTask implements PooledExchangeTask {
+ private static final AtomicIntegerFieldUpdater<RedeliveryTask> AWAITED
+ = AtomicIntegerFieldUpdater.newUpdater(RedeliveryTask.class,
"awaited");
+
// state
private Message original;
private Exchange exchange;
private AsyncCallback callback;
private int redeliveryCounter;
private long redeliveryDelay;
+ // the number of the first delivery whose callback is still to come,
or 0 when none. The callback and the
+ // takeover of a first delivery that throws before it calls the
callback compete for it (compareAndSet), so the
+ // exchange is continued exactly once. A field, not an object per
delivery, as the first delivery runs for every
+ // step of a route whose error handler can redeliver
+ private volatile int awaited;
+ // numbers the first deliveries of this task (it is reused when tasks
are pooled), so a callback of an earlier
+ // exchange never matches
+ private int deliveries;
// default behavior which can be overloaded on a per exception basis
private Predicate retryWhilePredicate;
@@ -1326,17 +1337,88 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
}
} else {
// Simple delivery
- processOutput(exchange, doneSync -> {
- // only continue with callback if we are done
- if (isDone(exchange)) {
- AsyncCallback cb = callback;
- taskFactory.release(this);
- reactiveExecutor.schedule(cb);
- } else {
- // error occurred so loop back around and call
ourselves
- reactiveExecutor.schedule(this);
- }
- });
+ deliver();
+ }
+ }
+
+ /**
+ * The first delivery. When the processor throws before it calls the
callback, such as an AsyncProcessor that
+ * throws from process(exchange, callback) instead of setting the
exception on the exchange, the error handler
+ * takes over that failure as when the processor sets it on the
exchange (redelivery, the dead letter channel
+ * and onException apply), as a redelivery that throws does.
+ */
+ private void deliver() {
+ // read while this task still owns it: once the callback is
called, the task may be released and reused
+ final Exchange delivered = exchange;
+ int number = ++deliveries;
+ if (number == 0) {
+ number = ++deliveries;
+ }
+ final int delivery = number;
+ // an ordered write is enough: the callback reaches another thread
only through a hand-off that publishes it
+ AWAITED.lazySet(this, delivery);
+ try {
+ processOutput(delivered, doneSync -> onDelivered(delivery));
+ } catch (Throwable e) {
+ // a Throwable, not only an Exception: an Error (such as
NoClassDefFoundError) thrown before the callback
+ // is called would also leave the exchange hanging, as the
reactive executor catches only exceptions
+ if (awaited != delivery) {
+ // the callback was called before the exception was
thrown, and it continued the exchange (this task
+ // may already be released), so only log the exception:
the exchange must not be continued again.
+ // This log does not read the exchange id: it is created
on first use, and the exchange may be in use
+ // on another thread (or, pooled, by another exchange)
+ LOG.warn("Error executing the first delivery by the
processor {} due to {}. The exchange continued, so"
+ + " this exception is ignored.",
+ outputAsync, e.getMessage(), e);
+ return;
+ }
+ // 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 (such
as the reply of a route behind a direct
+ // producer whose steps each schedule the next one there, as
with its default error handler) still
+ // comes first and continues the exchange. A callback that
only work queued after the takeover calls
+ // (such as a route with noErrorHandler()) loses to it
+ final String exchangeId = delivered.getExchangeId();
+ reactiveExecutor.schedule(() ->
takeOverFailedDelivery(delivery, delivered, exchangeId, e));
+ }
+ }
+
+ private void onDelivered(int delivery) {
+ if (!AWAITED.compareAndSet(this, delivery, 0)) {
+ // the processor threw before it called the callback, and the
error handler took over that failure, or
+ // the processor called the callback a second time
+ LOG.warn("The callback of the first delivery by the processor
{} was called when it was no longer awaited"
+ + " (after the processor had thrown an exception,
which the error handler handled instead, or"
+ + " more than once; an AsyncProcessor must not throw
from process). This callback is ignored.",
+ outputAsync);
+ return;
+ }
+ continueAfterDelivery();
+ }
+
+ private void takeOverFailedDelivery(int delivery, Exchange delivered,
String exchangeId, Throwable cause) {
+ if (AWAITED.compareAndSet(this, delivery, 0)) {
+ // a failure of the processor, so continue as when the
processor sets the exception on the exchange and
+ // calls the callback: the error handler handles it
(redeliver, onException or the dead letter channel)
+ delivered.setException(cause);
+ continueAfterDelivery();
+ } else {
+ // the queued callback continued the exchange (which may
already be done and released), so this
+ // exception is only logged
+ LOG.warn("Error executing the first delivery of exchangeId: {}
by the processor {} due to {}. The exchange"
+ + " continued, so this exception is ignored.",
+ exchangeId, outputAsync, cause.getMessage(), cause);
+ }
+ }
+
+ private void continueAfterDelivery() {
+ // only continue with callback if we are done
+ if (isDone(exchange)) {
+ AsyncCallback cb = callback;
+ taskFactory.release(this);
+ reactiveExecutor.schedule(cb);
+ } else {
+ // error occurred so loop back around and call ourselves
+ reactiveExecutor.schedule(this);
}
}
@@ -1462,10 +1544,12 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
final String exchangeId = redelivered.getExchangeId();
try {
doRedeliver(continued, exchangeId);
- } catch (Exception e) {
+ } catch (Throwable e) {
+ // a Throwable, not only an Exception, as in deliver()
if (continued.get()) {
// the callback was called before the exception was
thrown, and it continued the exchange (this task
- // may already be released), so leave the exception to the
reactive executor, which logs it, as before
+ // may already be released), so leave the exception to the
reactive executor, as before (a precise
+ // rethrow: only unchecked exceptions and errors reach
here)
throw e;
}
// the redelivery threw before the callback was called, such
as an AsyncProcessor that throws from
@@ -1479,7 +1563,7 @@ public abstract class RedeliveryErrorHandler extends
ErrorHandlerSupport
}
private void takeOverFailedRedelivery(
- AtomicBoolean continued, Exchange redelivered, String
exchangeId, Exception cause) {
+ AtomicBoolean continued, Exchange redelivered, String
exchangeId, Throwable cause) {
if (continued.compareAndSet(false, true)) {
// a new failure of the redelivery, so let the error handler
handle it (redeliver again or give up), as
// when the processor sets the exception on the exchange or
the onRedelivery processor fails
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsPooledTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsPooledTest.java
new file mode 100644
index 000000000000..51a8bc3fcad0
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsPooledTest.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.errorhandler;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.impl.engine.PooledProcessorExchangeFactory;
+
+/**
+ * The tests of {@link RedeliveryFirstDeliveryProcessorThrowsTestSupport} with
pooled exchanges, so the error handler
+ * uses pooled redelivery tasks, which are reset when they are released.
+ */
+class RedeliveryFirstDeliveryProcessorThrowsPooledTest extends
RedeliveryFirstDeliveryProcessorThrowsTestSupport {
+
+ @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/errorhandler/RedeliveryFirstDeliveryProcessorThrowsTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsTest.java
new file mode 100644
index 000000000000..cb57a79fa28c
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsTest.java
@@ -0,0 +1,407 @@
+/*
+ * 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.errorhandler;
+
+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.AsyncProcessor;
+import org.apache.camel.AsyncProducer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.AsyncProcessorSupport;
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * An {@link AsyncProcessor} that throws from {@code process(exchange,
callback)} on the first delivery, instead of
+ * setting the exception on the exchange, in a route whose error handler can
redeliver, has exception policies or a dead
+ * letter channel: the error handler must handle the thrown failure as one set
on the exchange (redelivery, the dead
+ * letter channel and onException apply, as for a redelivery that throws), and
continue the exchange exactly once. Each
+ * send is bounded, so a hang fails the test.
+ */
+class RedeliveryFirstDeliveryProcessorThrowsTest extends
RedeliveryFirstDeliveryProcessorThrowsTestSupport {
+
+ /**
+ * Throws on the first call, and succeeds on the later calls.
+ */
+ private 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;
+ }
+ }
+
+ /**
+ * Throws on the first call, and fails the exchange as an AsyncProcessor
should (the exception on the exchange) on
+ * the later calls.
+ */
+ private final class ThrowsThenFails extends AsyncProcessorSupport {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ int call = calls.incrementAndGet();
+ if (call == 1) {
+ throw new IllegalStateException("thrown");
+ }
+ exchange.setException(new IllegalStateException("failed " + call));
+ callback.done(true);
+ return true;
+ }
+ }
+
+ private void start(String errorHandler, Processor processor) throws
Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ switch (errorHandler) {
+ case "immediate" ->
errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0));
+ case "synchronousDelayed" ->
+
errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(10));
+ case "asynchronousDelayed" ->
errorHandler(defaultErrorHandler().maximumRedeliveries(1)
+ .redeliveryDelay(10).asyncDelayedRedelivery());
+ case "deadLetterChannel" ->
errorHandler(deadLetterChannel("mock:dead"));
+ case "deadLetterChannelRedelivery" ->
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(1).redeliveryDelay(0));
+ case "deadLetterChannelRedeliveryOnExceptionWithout" -> {
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(1).redeliveryDelay(0));
+
onException(IllegalStateException.class).maximumRedeliveries(0);
+ }
+ case "onException" ->
onException(IllegalStateException.class).handled(true).to("mock:handled");
+ default -> throw new
IllegalArgumentException(errorHandler);
+ }
+ from("direct:start").process(processor).to("mock:result");
+ }
+ });
+ context.start();
+ }
+
+ private void assertRedelivered(Exchange out) {
+ assertNull(out.getException());
+ assertEquals(2, calls.get(), "the processor was called for the
delivery and the redelivery");
+ assertEquals(1,
out.getMessage().getHeader(Exchange.REDELIVERY_COUNTER, Integer.class),
"redelivery counter");
+ assertEquals(1, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+
+ private void assertDeadLetterChannelGotTheThrownFailure(Exchange out,
MockEndpoint dead) throws Exception {
+ assertNull(out.getException(), "the dead letter channel handled the
failure");
+ dead.assertIsSatisfied();
+ assertEquals(1, dead.getReceivedCounter(), "mock:dead");
+ Exception caught =
dead.getReceivedExchanges().get(0).getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+ assertInstanceOf(IllegalStateException.class, caught);
+ assertEquals("thrown", caught.getMessage());
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @Test
+ void thrownFailureIsRedelivered() throws Exception {
+ start("immediate", new ThrowsFirst());
+ assertRedelivered(send());
+ }
+
+ @Test
+ void thrownFailureIsRedeliveredAfterASynchronousDelay() throws Exception {
+ start("synchronousDelayed", new ThrowsFirst());
+ assertRedelivered(send());
+ }
+
+ @Test
+ void thrownFailureIsRedeliveredAfterAnAsynchronousDelay() throws Exception
{
+ start("asynchronousDelayed", new ThrowsFirst());
+ assertRedelivered(send());
+ }
+
+ @Test
+ void transactedThrownFailureIsRedelivered() throws Exception {
+ // a transacted exchange runs on the reactive executor's queue, and
its redelivery is always synchronous
+ start("synchronousDelayed", new ThrowsFirst());
+ assertRedelivered(send(true));
+ }
+
+ @Test
+ void exhaustedFailsWithTheFailureOfTheRedelivery() throws Exception {
+ start("immediate", new ThrowsThenFails());
+ Exchange out = send();
+ assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals("failed 2", out.getException().getMessage());
+ assertEquals(2, calls.get(), "the processor was called for the
delivery and the redelivery");
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @Test
+ void throwsOnEveryAttempt() throws Exception {
+ // the redelivery that throws is handled as since CAMEL-26430: the
exchange fails with its exception
+ start("immediate", new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ throw new IllegalStateException("thrown " +
calls.incrementAndGet());
+ }
+ });
+ Exchange out = send();
+ assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals("thrown 2", out.getException().getMessage());
+ assertEquals(2, calls.get(), "the processor was called for the
delivery and the redelivery");
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @Test
+ void errorThrownOnEveryAttempt() throws Exception {
+ // an Error, not an Exception, thrown on the first delivery and on the
redelivery: before, the reactive executor
+ // let it through (it catches only exceptions), and the exchange never
completed
+ start("immediate", new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ throw new NoClassDefFoundError("thrown " +
calls.incrementAndGet());
+ }
+ });
+ Exchange out = send();
+ assertInstanceOf(NoClassDefFoundError.class,
out.getException().getCause());
+ assertEquals("thrown 2", out.getException().getCause().getMessage());
+ assertEquals(2, calls.get(), "the processor was called for the
delivery and the redelivery");
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @Test
+ void deadLetterChannelGetsTheFailureOfTheRedelivery() throws Exception {
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+ start("deadLetterChannelRedelivery", new ThrowsThenFails());
+ Exchange out = send();
+ assertNull(out.getException(), "the dead letter channel handled the
failure");
+ dead.assertIsSatisfied();
+ Exception caught =
dead.getReceivedExchanges().get(0).getProperty(Exchange.EXCEPTION_CAUGHT,
Exception.class);
+ assertEquals("failed 2", caught.getMessage());
+ assertEquals(2, calls.get(), "the processor was called for the
delivery and the redelivery");
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @Test
+ void notRedeliveredWithAnOnExceptionWithoutRedeliveries() throws Exception
{
+ // the way to not redeliver such a failure that the upgrade guide
gives: the dead letter channel still gets it
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+ start("deadLetterChannelRedeliveryOnExceptionWithout", new
ThrowsFirst());
+ assertDeadLetterChannelGotTheThrownFailure(send(), dead);
+ assertEquals(1, calls.get(), "calls");
+ }
+
+ @Test
+ void deadLetterChannelGetsTheThrownFailure() throws Exception {
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+ start("deadLetterChannel", new ThrowsFirst());
+ assertDeadLetterChannelGotTheThrownFailure(send(), dead);
+ assertEquals(1, calls.get(), "calls");
+ }
+
+ @Test
+ void onExceptionOfTheThrownFailureHandlesIt() throws Exception {
+ start("onException", new ThrowsFirst());
+ Exchange out = send();
+ assertNull(out.getException(), "the onException handled the thrown
failure");
+ assertEquals(1, getMockEndpoint("mock:handled").getReceivedCounter(),
"mock:handled");
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ assertEquals(1, calls.get(), "calls");
+ }
+
+ @Test
+ void nextExchangeIsProcessed() throws Exception {
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+ start("deadLetterChannel", new ThrowsFirst());
+ assertDeadLetterChannelGotTheThrownFailure(send(), dead);
+
+ Exchange out = send();
+ assertNull(out.getException());
+ assertEquals(1, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ assertEquals(2, calls.get(), "calls");
+ }
+
+ @Test
+ void controlFailureSetOnTheExchange() throws Exception {
+ // control, as before: a processor that sets the exception on the
exchange and calls the callback, as an
+ // AsyncProcessor should, goes to the dead letter channel
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+ start("deadLetterChannel", new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ exchange.setException(new IllegalStateException("thrown"));
+ callback.done(true);
+ return true;
+ }
+ });
+ assertDeadLetterChannelGotTheThrownFailure(send(), dead);
+ assertEquals(1, calls.get(), "calls");
+ }
+
+ @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, so the dead letter channel gets it
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+ start("deadLetterChannel", exchange -> {
+ calls.incrementAndGet();
+ throw new IllegalStateException("thrown");
+ });
+ assertDeadLetterChannelGotTheThrownFailure(send(), dead);
+ assertEquals(1, calls.get(), "calls");
+ }
+
+ @Test
+ void callbackOfARouteBeforeTheProcessorThrowsContinuesTheExchange() throws
Exception {
+ // the processor hands the exchange and the callback to a direct route
(with the default error handler) and then
+ // throws. Each step of that route runs on the reactive executor ahead
of the takeover queued there and the route
+ // calls the callback through it, so it continues the exchange: the
exchange completes with the reply of that
+ // route, once. (A route with noErrorHandler() would lose to the
takeover, and its callback be ignored.)
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:inner").setBody(constant("inner")).setHeader("inner",
constant("yes"));
+ }
+ });
+ AtomicReference<AsyncProducer> producer = new AtomicReference<>();
+ start("deadLetterChannel", new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ producer.get().process(exchange, callback);
+ throw new IllegalStateException("after the callback was handed
to a route");
+ }
+ });
+ AsyncProducer direct =
context.getEndpoint("direct:inner").createAsyncProducer();
+ direct.start();
+ producer.set(direct);
+ AtomicInteger done = countDoneExchanges();
+ try {
+ Exchange out = sendOnThisThread();
+ assertNull(out.getException());
+ assertEquals("inner", out.getMessage().getBody());
+ assertEquals("yes", out.getMessage().getHeader("inner"));
+ assertEquals(1,
getMockEndpoint("mock:result").getReceivedCounter(), "mock:result");
+ assertEquals(0, getMockEndpoint("mock:dead").getReceivedCounter(),
"mock:dead");
+ assertEquals(1, done.get(), "the exchange was done once");
+ assertEquals(1, calls.get(), "calls");
+ } finally {
+ direct.stop();
+ }
+ }
+
+ @Test
+ void callbackCalledAfterTheProcessorThrewIsIgnored() throws Exception {
+ // the processor hands the callback to another thread and then throws.
The error handler cannot tell this from a
+ // callback that never comes, so it takes over the thrown failure, 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<>();
+ MockEndpoint dead = getMockEndpoint("mock:dead");
+ dead.expectedMessageCount(1);
+ AtomicInteger done = countDoneExchanges();
+ start("deadLetterChannel", 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");
+ }
+ });
+
+ assertDeadLetterChannelGotTheThrownFailure(send(), dead);
+
+ // 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(1, dead.getReceivedCounter(), "mock:dead");
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ assertEquals(1, 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
(to mock:result, or with the thrown
+ // failure to the dead letter channel)
+ int count = 50;
+ ExecutorService callbacks = Executors.newFixedThreadPool(2);
+ try {
+ start("deadLetterChannel", 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("direct:start", false));
+ }
+ for (CompletableFuture<Exchange> send : sends) {
+ assertNull(send.get(20, TimeUnit.SECONDS).getException());
+ }
+ 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(count,
getMockEndpoint("mock:result").getReceivedCounter()
+ +
getMockEndpoint("mock:dead").getReceivedCounter(),
+ "mock:result + mock:dead");
+ assertEquals(count, calls.get(), "calls");
+ } finally {
+ callbacks.shutdownNow();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsTestSupport.java
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsTestSupport.java
new file mode 100644
index 000000000000..61fef9f2f378
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RedeliveryFirstDeliveryProcessorThrowsTestSupport.java
@@ -0,0 +1,78 @@
+/*
+ * 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.errorhandler;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+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.assertNull;
+
+/**
+ * Shared by {@link RedeliveryFirstDeliveryProcessorThrowsTest} and
+ * {@link RedeliveryFirstDeliveryProcessorThrowsPooledTest}: the tests that
depend on whether the redelivery tasks are
+ * pooled. The route has an onException, so its error handler runs the step
with its redelivery task, and the tests
+ * inherited from {@link SimpleTaskProcessorThrowsTestSupport} (the late
callback with the next exchange, the double
+ * callback) run on the first delivery of that task.
+ */
+abstract class RedeliveryFirstDeliveryProcessorThrowsTestSupport extends
SimpleTaskProcessorThrowsTestSupport {
+
+ @Override
+ protected void start(Processor processor) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // not handled and no redelivery: the exchange fails with the
thrown exception, as with the simple task
+
onException(IllegalStateException.class).maximumRedeliveries(0);
+ from("direct:start").process(processor).to("mock:result");
+ }
+ });
+ context.start();
+ }
+
+ @Test
+ void callbackCalledBeforeTheProcessorThrowsContinuesTheExchangeOnce()
throws Exception {
+ // the processor calls the callback and then throws: the callback
continued the exchange, so the exception is
+ // only logged, and the exchange completes once (before, the catch of
the task failed the exchange and completed
+ // it a second time)
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(1);
+ start(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();
+ assertNull(out.getException());
+ result.assertIsSatisfied();
+ assertEquals(1, result.getReceivedCounter(), "mock:result");
+ assertEquals(1, done.get(), "the exchange was done once");
+ assertEquals(1, calls.get(), "calls");
+ }
+}
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 5d88bf55901e..ecaafe8c6090 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
@@ -206,6 +206,25 @@ slip) did not send its message. A route called from such a
`doFinally` that fail
own error handler, as anywhere else; before, its error handler left the
failure alone and the exchange failed with that
exception. As before, the route does not continue after the `doTry` once a
called route handled the failure.
+=== camel-core - an AsyncProcessor that throws on the first attempt goes
through the error handler
+
+When the error handler of a step can redeliver, has an `onException` or an
`onPrepareFailure` processor, or is a dead
+letter channel, and the processor of the step (such as a custom
`AsyncProcessor` used with `process(...)`; an endpoint
+in `to(...)` is not affected) throws from `process(exchange, callback)` on the
first attempt instead of setting the
+exception on the exchange, the error handler now handles that failure as any
other failure of the step: it redelivers
+it, and when the redeliveries are exhausted, the matching `onException` or the
dead letter channel handles it. This is
+what it already did when the processor set the same exception on the exchange,
or when a plain `Processor` threw it.
+Before, the exchange failed at once with the thrown exception. When such a
processor calls its callback and then
+throws, the exception is now only logged at WARN; when it handed its callback
to another thread before it threw, and
+that thread calls it after the error handler took over the failure, or when it
calls its callback twice, that call is
+now ignored and logged at WARN. Before, the exchange was continued a second
time in these cases.
+
+To not redeliver such a failure, add an `onException` for its exception type
with `maximumRedeliveries(0)`; as for the
+same failure set on the exchange, the dead letter channel and the
`onException` still handle it. There is no option to
+let such a failure skip the dead letter channel or `onException` as before; a
plain `Processor` that throws gets the
+same handling. Better, fix the processor to set the exception on the exchange
and call the callback, as an
+`AsyncProcessor` must.
+
=== camel-core - ThrottlingExceptionRoutePolicy counts the failures within the
failure window
The circuit of `ThrottlingExceptionRoutePolicy` now opens when
`failureThreshold` failures occurred within the last
