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 9e4f22d16509 CAMEL-26882: camel-core - DefaultProducerCache: an
AsyncProducer that calls the callback and then throws completes the exchange
once (#29786)
9e4f22d16509 is described below
commit 9e4f22d16509c6e5cc7fb22332697d35676c2060
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 15:01:53 2026 +0530
CAMEL-26882: camel-core - DefaultProducerCache: an AsyncProducer that calls
the callback and then throws completes the exchange once (#29786)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/camel/spi/ProducerCache.java | 3 +-
.../DefaultProducerCacheProducerThrowsTest.java | 269 +++++++++++++++++++++
.../camel/support/cache/DefaultProducerCache.java | 101 ++++++--
.../camel/support/cache/EventNotifierCallback.java | 14 +-
4 files changed, 361 insertions(+), 26 deletions(-)
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
b/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
index 74971d0e2cb5..cd5036d6a552 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/ProducerCache.java
@@ -139,7 +139,8 @@ public interface ProducerCache extends Service {
/**
* Sends an exchange to an endpoint using a supplied callback supporting
the asynchronous routing engine.
* <p/>
- * If an exception was thrown during processing, it would be set on the
given Exchange
+ * If an exception was thrown during processing, it would be set on the
given Exchange, unless the producer had
+ * already completed the exchange (called the callback) before it threw:
then the callback is not called again.
*
* @param endpoint the endpoint to send the exchange to
* @param exchange the exchange, can be <tt>null</tt> if so then
create a new exchange from the producer
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheProducerThrowsTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheProducerThrowsTest.java
new file mode 100644
index 000000000000..319bcbb45dde
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheProducerThrowsTest.java
@@ -0,0 +1,269 @@
+/*
+ * 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.impl;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProducer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+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.spi.CamelEvent;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.EventNotifierSupport;
+import org.apache.camel.support.cache.DefaultProducerCache;
+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.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * An AsyncProducer that breaks the async contract by throwing from process,
before or after it called the callback,
+ * sent through the producer cache.
+ */
+public class DefaultProducerCacheProducerThrowsTest extends ContextTestSupport
{
+
+ private final AtomicInteger created = new AtomicInteger();
+ private final AtomicInteger sending = new AtomicInteger();
+ private final AtomicInteger sent = new AtomicInteger();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof CamelEvent.ExchangeSendingEvent) {
+ sending.incrementAndGet();
+ } else if (event instanceof CamelEvent.ExchangeSentEvent) {
+ sent.incrementAndGet();
+ }
+ }
+ });
+ return camelContext;
+ }
+
+ @Test
+ public void testCallbackThenThrowCompletesOnce() {
+ DefaultProducerCache cache = new DefaultProducerCache(this, context,
0);
+ cache.start();
+ try {
+ Exchange exchange = newExchange("after");
+ AtomicInteger calls = new AtomicInteger();
+
+ boolean sync = cache.doInAsyncProducer(endpoint(), exchange,
doneSync -> calls.incrementAndGet(),
+ (producer, e, callback) -> producer.process(e, callback));
+
+ assertTrue(sync);
+ assertEquals(1, calls.get(), "The callback must be called once");
+ assertNull(exchange.getException(), "The producer completed the
exchange before it threw");
+ } finally {
+ cache.stop();
+ }
+ }
+
+ @Test
+ public void testThrowBeforeCallbackFailsAndReleasesProducer() {
+ DefaultProducerCache cache = new DefaultProducerCache(this, context,
0);
+ cache.start();
+ try {
+ Endpoint endpoint = endpoint();
+ for (int i = 0; i < 2; i++) {
+ Exchange exchange = newExchange("before");
+ AtomicInteger calls = new AtomicInteger();
+
+ boolean sync = cache.doInAsyncProducer(endpoint, exchange,
doneSync -> calls.incrementAndGet(),
+ (producer, e, callback) -> producer.process(e,
callback));
+
+ assertTrue(sync);
+ assertEquals(1, calls.get(), "The callback must be called
once");
+ assertInstanceOf(IllegalStateException.class,
exchange.getException());
+ }
+ // the producer (of an endpoint that is not a singleton producer)
is back in the pool after each send
+ assertEquals(1, created.get());
+ } finally {
+ cache.stop();
+ }
+ }
+
+ @Test
+ public void testErrorBeforeCallbackFailsAndReleasesProducer() {
+ DefaultProducerCache cache = new DefaultProducerCache(this, context,
0);
+ cache.start();
+ try {
+ Endpoint endpoint = endpoint();
+ for (int i = 0; i < 2; i++) {
+ Exchange exchange = newExchange("error");
+ AtomicInteger calls = new AtomicInteger();
+
+ boolean sync = cache.doInAsyncProducer(endpoint, exchange,
doneSync -> calls.incrementAndGet(),
+ (producer, e, callback) -> producer.process(e,
callback));
+
+ assertTrue(sync);
+ assertEquals(1, calls.get(), "The callback must be called
once");
+ // the exchange keeps an Error wrapped, as for any Error
+ assertInstanceOf(NoClassDefFoundError.class,
exchange.getException().getCause());
+ }
+ // the producer is back in the pool after each send
+ assertEquals(1, created.get());
+ } finally {
+ cache.stop();
+ }
+ }
+
+ @Test
+ public void testAsyncSendCallbackThenThrowKeepsTheResult() throws
Exception {
+ DefaultProducerCache cache = new DefaultProducerCache(this, context,
0);
+ cache.start();
+ try {
+ Exchange exchange = newExchange("after");
+
+ CompletableFuture<Exchange> future =
cache.asyncSendExchange(endpoint(), null, null, null, exchange, null);
+
+ assertTrue(future.isDone());
+ assertNull(future.get(5, TimeUnit.SECONDS).getException(),
+ "The producer completed the exchange before it threw");
+ assertEquals("Done", exchange.getMessage().getBody());
+ } finally {
+ cache.stop();
+ }
+ }
+
+ @Test
+ public void testAsyncSendCallbackThenThrowReleasesTheProducerOnce() throws
Exception {
+ DefaultProducerCache cache = new DefaultProducerCache(this, context,
0);
+ cache.start();
+ try {
+ Endpoint endpoint = endpoint();
+ cache.asyncSendExchange(endpoint, null, null, null,
newExchange("after"), null).get(5, TimeUnit.SECONDS);
+
+ // the producer is in the pool once: two sends at the same time
get two producers
+ AsyncProducer first = cache.acquireProducer(endpoint);
+ AsyncProducer second = cache.acquireProducer(endpoint);
+ assertNotSame(first, second);
+ cache.releaseProducer(endpoint, first);
+ cache.releaseProducer(endpoint, second);
+ } finally {
+ cache.stop();
+ }
+ }
+
+ @Test
+ public void testAsyncSendThrowBeforeCallbackPairsTheEvents() throws
Exception {
+ DefaultProducerCache cache = new DefaultProducerCache(this, context,
0);
+ cache.setEventNotifierEnabled(true);
+ cache.start();
+ try {
+ Exchange exchange = newExchange("before");
+ cache.asyncSendExchange(endpoint(), null, null, null, exchange,
null).get(5, TimeUnit.SECONDS);
+
+ assertInstanceOf(IllegalStateException.class,
exchange.getException());
+ assertTrue(sending.get() > 0);
+ assertEquals(sending.get(), sent.get(), "each sending event has
its sent event");
+ } finally {
+ cache.stop();
+ }
+ }
+
+ @Test
+ public void testToDCallbackThenThrowRoutesOnce() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("Done");
+
+ Exchange out = template.send("direct:start", e -> {
+ e.getMessage().setHeader("mode", "after");
+ e.getMessage().setHeader("target", "producer");
+ });
+
+ // the rest of the route runs once, and without the exception the
producer threw after it completed
+ mock.assertIsSatisfied();
+ assertEquals(1, mock.getReceivedCounter());
+ assertNull(out.getException());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ context.addEndpoint("throwing:producer", new
ThrowingEndpoint());
+
from("direct:start").toD("throwing:${header.target}").to("mock:result");
+ }
+ };
+ }
+
+ private Endpoint endpoint() {
+ return context.getEndpoint("throwing:producer");
+ }
+
+ private Exchange newExchange(String mode) {
+ Exchange exchange = endpoint().createExchange();
+ exchange.getMessage().setHeader("mode", mode);
+ return exchange;
+ }
+
+ private final class ThrowingEndpoint extends DefaultEndpoint {
+
+ ThrowingEndpoint() {
+ super("throwing:producer", null);
+ setCamelContext(context);
+ }
+
+ @Override
+ public boolean isSingletonProducer() {
+ return false;
+ }
+
+ @Override
+ public Producer createProducer() {
+ created.incrementAndGet();
+ return new DefaultAsyncProducer(this) {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback
callback) {
+ Object mode = exchange.getMessage().getHeader("mode");
+ if ("error".equals(mode)) {
+ // such as a class of the producer that is missing
+ throw new NoClassDefFoundError("Simulated missing
class");
+ }
+ if ("after".equals(mode)) {
+ exchange.getMessage().setBody("Done");
+ callback.done(true);
+ }
+ throw new IllegalStateException("Simulated producer that
throws");
+ }
+ };
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
b/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
index 4eaec5a6bd46..3b476bd99bc6 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/cache/DefaultProducerCache.java
@@ -18,6 +18,7 @@ package org.apache.camel.support.cache;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
@@ -259,6 +260,7 @@ public class DefaultProducerCache extends ServiceSupport
implements ProducerCach
return true;
}
+ SentCallback sent = null;
try {
// emit event that the exchange is being sent, and record timing
for sending the exchange using the producer
final StopWatch watch;
@@ -269,27 +271,74 @@ public class DefaultProducerCache extends ServiceSupport
implements ProducerCach
}
// invoke the callback, which calls the producer (with the send
hook around it if any)
- return
producerCallback.doInAsyncProducer(SendHelper.withSendHook(camelContext,
producer), exchange, doneSync -> {
- try {
- // emit event that the exchange was sent to the endpoint
- SendHelper.notifySent(exchange, endpoint, watch);
-
- // release back to the pool
- releaseProducer(endpoint, producer);
- } finally {
- callback.done(doneSync);
- }
- });
- } catch (Exception e) {
+ sent = new SentCallback(endpoint, exchange, producer, watch,
callback);
+ return
producerCallback.doInAsyncProducer(SendHelper.withSendHook(camelContext,
producer), exchange, sent);
+ } catch (Throwable e) {
+ if (sent != null && !sent.compareAndSet(false, true)) {
+ // the producer called the callback and then threw: the
exchange is already continued (maybe on
+ // another thread), so it is not completed a second time and
the exception is not set on it
+ LOG.warn("The producer of {} threw an exception after it
completed the exchange. This exception is ignored.",
+ endpoint, e);
+ return sent.doneSync;
+ }
// ensure exceptions is caught and set on the exchange
- if (exchange != null) {
- exchange.setException(e);
+ exchange.setException(e);
+ if (sent != null) {
+ // the producer threw before it called the callback: complete
the send as if it had called it
+ sent.complete(true);
+ } else {
+ callback.done(true);
}
- callback.done(true);
return true;
}
}
+ /**
+ * The callback of a send in {@link #doInAsyncProducer}: emits the sent
event, releases the producer and calls the
+ * callback of the caller, once (also when the producer calls its callback
and then throws).
+ */
+ private final class SentCallback extends AtomicBoolean implements
AsyncCallback {
+ private final Endpoint endpoint;
+ private final Exchange exchange;
+ private final AsyncProducer producer;
+ private final StopWatch watch;
+ private final AsyncCallback callback;
+ private volatile boolean doneSync;
+
+ private SentCallback(Endpoint endpoint, Exchange exchange,
AsyncProducer producer, StopWatch watch,
+ AsyncCallback callback) {
+ this.endpoint = endpoint;
+ this.exchange = exchange;
+ this.producer = producer;
+ this.watch = watch;
+ this.callback = callback;
+ }
+
+ @Override
+ public void done(boolean doneSync) {
+ if (compareAndSet(false, true)) {
+ complete(doneSync);
+ } else {
+ LOG.warn("The producer of {} completed the exchange after the
send was completed (more than once, or after"
+ + " it threw). This is ignored.",
+ endpoint);
+ }
+ }
+
+ private void complete(boolean doneSync) {
+ this.doneSync = doneSync;
+ try {
+ // emit event that the exchange was sent to the endpoint
+ SendHelper.notifySent(exchange, endpoint, watch);
+
+ // release back to the pool
+ releaseProducer(endpoint, producer);
+ } finally {
+ callback.done(doneSync);
+ }
+ }
+ }
+
protected boolean asyncDispatchExchange(
Endpoint endpoint, AsyncProducer producer,
Processor resultProcessor, Exchange exchange, AsyncCallback
callback) {
@@ -299,18 +348,22 @@ public class DefaultProducerCache extends ServiceSupport
implements ProducerCach
// set property which endpoint we send to
exchange.setProperty(ExchangePropertyKey.TO_ENDPOINT,
endpoint.getEndpointUri());
- // send the exchange using the processor
+ // send the exchange using the processor; an exception thrown by the
producer is handled by doInAsyncProducer,
+ // which knows whether the producer called the callback before it threw
+ EventNotifierCallback notifier = null;
+ if (eventNotifierEnabled &&
camelContext.getCamelContextExtension().isEventNotificationApplicable()) {
+ notifier = new EventNotifierCallback(callback, exchange, endpoint);
+ callback = notifier;
+ }
try {
- if (eventNotifierEnabled &&
camelContext.getCamelContextExtension().isEventNotificationApplicable()) {
- callback = new EventNotifierCallback(callback, exchange,
endpoint);
- }
// invoke the asynchronous method
return sharedInternalProcessor.process(exchange, callback,
producer, resultProcessor);
- } catch (Exception e) {
- // ensure exceptions is caught and set on the exchange
- exchange.setException(e);
- callback.done(true);
- return true;
+ } catch (Throwable e) {
+ if (notifier != null) {
+ // pair the sending event with a sent event, unless the
callback did already
+ notifier.notifySentOnce();
+ }
+ throw e;
}
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
b/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
index 38cb09ae0494..6a5b834f9a23 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/cache/EventNotifierCallback.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.support.cache;
+import java.util.concurrent.atomic.AtomicBoolean;
+
import org.apache.camel.AsyncCallback;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
@@ -30,6 +32,7 @@ class EventNotifierCallback implements AsyncCallback {
private final StopWatch watch;
private final Exchange exchange;
private final Endpoint endpoint;
+ private final AtomicBoolean sent = new AtomicBoolean();
public EventNotifierCallback(AsyncCallback originalCallback, Exchange
exchange,
Endpoint endpoint) {
@@ -41,7 +44,16 @@ class EventNotifierCallback implements AsyncCallback {
@Override
public void done(boolean doneSync) {
- SendHelper.notifySent(exchange, endpoint, watch);
+ notifySentOnce();
originalCallback.done(doneSync);
}
+
+ /**
+ * Emits the sent event, once (the producer may throw after it called the
callback).
+ */
+ void notifySentOnce() {
+ if (sent.compareAndSet(false, true)) {
+ SendHelper.notifySent(exchange, endpoint, watch);
+ }
+ }
}
