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 0c158cbdb6f7 CAMEL-26461: camel-core - NoErrorHandler: an
AsyncProcessor that throws fails the exchange instead of hanging (#29041)
0c158cbdb6f7 is described below
commit 0c158cbdb6f77e826782695164ba468bd8f753c5
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 3 16:08:54 2026 +0530
CAMEL-26461: camel-core - NoErrorHandler: an AsyncProcessor that throws
fails the exchange instead of hanging (#29041)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../processor/errorhandler/NoErrorHandler.java | 80 +++-
.../NoErrorHandlerProcessorThrowsTest.java | 418 +++++++++++++++++++++
2 files changed, 493 insertions(+), 5 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/NoErrorHandler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/NoErrorHandler.java
index 318615028678..fae0dfb60b61 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/NoErrorHandler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/NoErrorHandler.java
@@ -17,6 +17,7 @@
package org.apache.camel.processor.errorhandler;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
@@ -27,9 +28,13 @@ import
org.apache.camel.support.AsyncCallbackToCompletableFutureAdapter;
import org.apache.camel.support.AsyncProcessorConverterHelper;
import org.apache.camel.support.AsyncProcessorHelper;
import org.apache.camel.support.service.ServiceHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class NoErrorHandler extends ErrorHandlerSupport implements
AsyncProcessor {
+ private static final Logger LOG =
LoggerFactory.getLogger(NoErrorHandler.class);
+
private final AsyncProcessor output;
public NoErrorHandler(Processor processor) {
@@ -43,13 +48,78 @@ public class NoErrorHandler extends ErrorHandlerSupport
implements AsyncProcesso
@Override
public boolean process(Exchange exchange, final AsyncCallback callback) {
- return output.process(exchange, new AsyncCallback() {
- @Override
- public void done(boolean doneSync) {
+ final Continuation continuation = new Continuation(output, exchange,
callback);
+ try {
+ return output.process(exchange, continuation);
+ } 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 (continuation.continued != 0) {
+ // the callback was called before the exception was thrown,
and it continued the exchange: leave the
+ // exception to the caller, as before (a precise rethrow: only
unchecked exceptions and errors reach here)
+ 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: the exception would be
+ // dropped by the reactive executor and the exchange would never
complete. Fail the exchange 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
+ final String exchangeId = exchange.getExchangeId();
+
exchange.getContext().getCamelContextExtension().getReactiveExecutor()
+ .schedule(() -> continuation.takeOver(exchangeId, e));
+ return false;
+ }
+ }
+
+ /**
+ * The callback given to the processor. It and the takeover of a processor
that throws before it calls the callback
+ * compete for continuing the exchange (compareAndSet), so the exchange is
continued exactly once.
+ */
+ private static final class Continuation implements AsyncCallback {
+ private static final AtomicIntegerFieldUpdater<Continuation> CONTINUED
+ = AtomicIntegerFieldUpdater.newUpdater(Continuation.class,
"continued");
+
+ private final AsyncProcessor output;
+ private final Exchange exchange;
+ private final AsyncCallback callback;
+ // 1 once the exchange is continued (a field, not an object, as this
callback is created for every step)
+ private volatile int continued;
+
+ Continuation(AsyncProcessor output, Exchange exchange, AsyncCallback
callback) {
+ this.output = output;
+ this.exchange = exchange;
+ this.callback = callback;
+ }
+
+ @Override
+ public void done(boolean doneSync) {
+ if (!CONTINUED.compareAndSet(this, 0, 1)) {
+ // this log does not read the exchange id: it is created on
first use, and the exchange may be in use
+ // on another thread
+ LOG.warn("The callback of {} 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.",
+ output);
+ return;
+ }
+ exchange.getExchangeExtension().setRedeliveryExhausted(false);
+ callback.done(doneSync);
+ }
+
+ void takeOver(String exchangeId, Throwable cause) {
+ if (CONTINUED.compareAndSet(this, 0, 1)) {
+ // fail the exchange, as when the processor sets the exception
on the exchange
+ exchange.setException(cause);
exchange.getExchangeExtension().setRedeliveryExhausted(false);
- callback.done(doneSync);
+ callback.done(false);
+ } else {
+ // the queued callback continued the exchange (which may
already be done), so this exception is only
+ // logged
+ LOG.warn("Error processing exchangeId: {} by {} due to {}. The
exchange continued, so this exception is"
+ + " ignored.",
+ exchangeId, output, cause.getMessage(), cause);
}
- });
+ }
}
@Override
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/NoErrorHandlerProcessorThrowsTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/NoErrorHandlerProcessorThrowsTest.java
new file mode 100644
index 000000000000..7ed348983700
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/NoErrorHandlerProcessorThrowsTest.java
@@ -0,0 +1,418 @@
+/*
+ * 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.Map;
+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.CamelExchangeException;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.AsyncProcessorSupport;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.support.DefaultEndpoint;
+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)}, instead of setting the exception on
+ * the exchange, in a route with {@code noErrorHandler()}: the exchange must
complete with the thrown exception instead
+ * of hanging, and be continued exactly once. Each send is bounded, so a hang
fails the test. The helpers, and the tests
+ * of a late callback with the next exchange and of a double callback, are
those of
+ * {@link SimpleTaskProcessorThrowsTestSupport}, with the route of {@link
#start(Processor)}.
+ */
+class NoErrorHandlerProcessorThrowsTest extends
SimpleTaskProcessorThrowsTestSupport {
+
+ /**
+ * 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;
+ }
+ }
+
+ @Override
+ protected void start(Processor processor) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").errorHandler(noErrorHandler()).process(processor).to("mock:result");
+ }
+ });
+ context.start();
+ }
+
+ @Test
+ void noErrorHandler() throws Exception {
+ start(new ThrowsFirst());
+ assertFailedWithTheThrownException(send());
+ }
+
+ @Test
+ void processorThrowsError() throws Exception {
+ // an Error, not an Exception: before, the reactive executor let it
through (it catches only exceptions), and the
+ // exchange never completed
+ start(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:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @Test
+ void afterAnAsynchronousStep() throws Exception {
+ // the step runs on the thread of the asynchronous delay, not on the
caller's
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").errorHandler(noErrorHandler()).delay(10).asyncDelayed().process(new
ThrowsFirst())
+ .to("mock:result");
+ }
+ });
+ context.start();
+ assertFailedWithTheThrownException(send());
+ }
+
+ @Test
+ void controlTransacted() throws Exception {
+ // control: a transacted exchange runs on the reactive executor's
queue, and the exchange fails with the thrown
+ // exception on main too
+ start(new ThrowsFirst());
+ assertFailedWithTheThrownException(send(true));
+ }
+
+ @Test
+ void nextExchangeIsProcessed() throws Exception {
+ start(new ThrowsFirst());
+ assertFailedWithTheThrownException(send());
+
+ Exchange out = send();
+ assertNull(out.getException());
+ assertEquals(1, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ assertEquals(2, 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
+ start(exchange -> {
+ calls.incrementAndGet();
+ throw new IllegalStateException("thrown");
+ });
+ assertFailedWithTheThrownException(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 reactive executor (which logs it), and
the exchange completes once
+ 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");
+ }
+
+ @Test
+ void controlTransactedCallbackCalledBeforeTheProcessorThrows() throws
Exception {
+ // control, as before: for a transacted exchange the exception thrown
after the callback reaches the caller, which
+ // fails the exchange (so a transaction rolls back)
+ start(new AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ calls.incrementAndGet();
+ callback.done(true);
+ throw new IllegalStateException("after the callback");
+ }
+ });
+
+ Exchange out = send(true);
+ assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals("after the callback", out.getException().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(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();
+ assertNull(out.getException());
+ assertEquals("queued", out.getMessage().getBody());
+ assertEquals(1, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ assertEquals(1, done.get(), "the exchange was done once");
+ assertEquals(1, calls.get(), "calls");
+ }
+
+ @Test
+ void controlCallbackOfARouteBeforeTheProcessorThrows() throws Exception {
+ // control, as before: 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 ahead
of the takeover queued on the reactive
+ // executor, and the route calls the callback through it, so the route
continues the exchange: the exchange
+ // completes with the reply of that route, once. (A called route with
noErrorHandler() loses to the takeover,
+ // even with one step: its steps do not go through the reactive
executor ahead of the takeover; the exchange
+ // then fails with the thrown exception and the reply of that route is
ignored with a WARN.)
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:inner").setBody(constant("inner")).setHeader("inner",
constant("yes"));
+ }
+ });
+ AtomicReference<AsyncProducer> producer = new AtomicReference<>();
+ start(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(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.
NoErrorHandler cannot tell this from a
+ // callback that never comes, so it fails the exchange with the thrown
exception (before, the exchange waited
+ // for the callback), 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<>();
+ AtomicInteger done = countDoneExchanges();
+ start(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");
+ }
+ });
+
+ assertFailedWithTheThrownException(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:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @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 with the thrown failure)
+ int count = 50;
+ ExecutorService callbacks = Executors.newFixedThreadPool(2);
+ try {
+ start(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));
+ }
+ int succeeded = 0;
+ for (CompletableFuture<Exchange> send : sends) {
+ Exchange out = send.get(20, TimeUnit.SECONDS);
+ if (out.getException() == null) {
+ succeeded++;
+ } else {
+ assertEquals("thrown", out.getException().getMessage());
+ }
+ }
+ 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:result").getReceivedCounter(), "mock:result");
+ assertEquals(count, calls.get(), "calls");
+ } finally {
+ callbacks.shutdownNow();
+ }
+ }
+
+ /**
+ * An endpoint whose producer is an AsyncProcessor that throws from
process(exchange, callback).
+ */
+ private void addThrowingEndpoint() {
+ context.addComponent("throwing", new DefaultComponent() {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ return new DefaultEndpoint(uri, this) {
+ @Override
+ public Producer createProducer() {
+ return new DefaultAsyncProducer(this) {
+ @Override
+ public boolean process(Exchange exchange,
AsyncCallback callback) {
+ calls.incrementAndGet();
+ throw new IllegalStateException("thrown");
+ }
+ };
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ });
+ }
+
+ @Test
+ void recipientListWithARecipientThatThrows() throws Exception {
+ // a recipient list wraps each producer in a clone of the error
handler of the route: the recipient that throws
+ // fails, and without stopOnException the next recipient still gets
the message
+ addThrowingEndpoint();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").errorHandler(noErrorHandler()).recipientList(constant("throwing:a,mock:b")).end()
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ Exchange out = send();
+ assertEquals(1, calls.get(), "calls");
+ assertEquals(1, getMockEndpoint("mock:b").getReceivedCounter(),
"mock:b");
+ // the failure of the recipient fails the exchange at the end of the
recipient list
+ assertInstanceOf(IllegalStateException.class, out.getException());
+ assertEquals("thrown", out.getException().getMessage());
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+
+ @Test
+ void recipientListStopOnExceptionWithARecipientThatThrows() throws
Exception {
+ addThrowingEndpoint();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").errorHandler(noErrorHandler()).recipientList(constant("throwing:a,mock:b"))
+ .stopOnException().end()
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ Exchange out = send();
+ assertInstanceOf(CamelExchangeException.class, out.getException());
+ assertInstanceOf(IllegalStateException.class,
out.getException().getCause());
+ assertEquals(1, calls.get(), "calls");
+ assertEquals(0, getMockEndpoint("mock:b").getReceivedCounter(),
"mock:b");
+ assertEquals(0, getMockEndpoint("mock:result").getReceivedCounter(),
"mock:result");
+ }
+}
