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 8c0259cd497c CAMEL-26458: camel-disruptor - multipleConsumers:
complete the exchange once when a consumer fails (#29038)
8c0259cd497c is described below
commit 8c0259cd497c8378243eef078c63c75262766016
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 3 13:25:46 2026 +0530
CAMEL-26458: camel-disruptor - multipleConsumers: complete the exchange
once when a consumer fails (#29038)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../MultipleConsumerSynchronizedExchange.java | 26 +++-
...DisruptorMultipleConsumersOnCompletionTest.java | 161 +++++++++++++++++++++
.../MultipleConsumerSynchronizedExchangeTest.java | 103 +++++++++++++
3 files changed, 282 insertions(+), 8 deletions(-)
diff --git
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/MultipleConsumerSynchronizedExchange.java
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/MultipleConsumerSynchronizedExchange.java
index 2825206fbb38..b2f2df6abd35 100644
---
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/MultipleConsumerSynchronizedExchange.java
+++
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/MultipleConsumerSynchronizedExchange.java
@@ -34,6 +34,8 @@ public class MultipleConsumerSynchronizedExchange extends
AbstractSynchronizedEx
private final AtomicBoolean resultHandled = new AtomicBoolean();
+ private final AtomicBoolean failed = new AtomicBoolean();
+
public MultipleConsumerSynchronizedExchange(Exchange exchange, int
expectedConsumers) {
super(exchange);
this.expectedConsumers = expectedConsumers;
@@ -42,17 +44,25 @@ public class MultipleConsumerSynchronizedExchange extends
AbstractSynchronizedEx
@Override
public void consumed(Exchange result) {
+ boolean failure = result.getException() != null;
+ if (failure) {
+ // before this consumer is counted, so that the consumer that is
counted last knows of the failure: this
+ // write comes before our incrementAndGet in program order, and
the last consumer's incrementAndGet on the
+ // same AtomicInteger happens after ours, so its failed.get() sees
it
+ failed.set(true);
+ }
+ int processed = processedConsumers.incrementAndGet();
- if (processedConsumers.incrementAndGet() == expectedConsumers ||
result.getException() != null
- && !resultHandled.getAndSet(true)) {
- // all consumers are done processing or an exception occurred
-
- //SEDA Does not configure an aggregator in the internally used
MulticastProcessor
- //As a result, the behaviour of SEDA in case of multicast is to
only copy the results on an error
- if (result.getException() != null) {
+ // complete the exchange once: at the first failure (the consumers
that are still routing then complete
+ // nothing), or when all consumers are done and none of them failed
+ if (failure) {
+ if (!resultHandled.getAndSet(true)) {
+ // only the result of a failure is copied back (as SEDA, whose
multicast to its consumers has no
+ // aggregation strategy)
ExchangeHelper.copyResults(getExchange(), result);
+ performSynchronization();
}
-
+ } else if (processed == expectedConsumers && !failed.get() &&
!resultHandled.getAndSet(true)) {
performSynchronization();
}
}
diff --git
a/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorMultipleConsumersOnCompletionTest.java
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorMultipleConsumersOnCompletionTest.java
new file mode 100644
index 000000000000..2c599456e071
--- /dev/null
+++
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorMultipleConsumersOnCompletionTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.component.disruptor;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With multipleConsumers the on completions of an InOnly exchange are handed
over to the consumers, and must run once,
+ * when one of the consumers fails or when all of them are done.
+ */
+public class DisruptorMultipleConsumersOnCompletionTest extends
CamelTestSupport {
+
+ private final List<String> events = new CopyOnWriteArrayList<>();
+ private final CountDownLatch release = new CountDownLatch(1);
+ private NotifyBuilder consumersDone;
+
+ @AfterEach
+ void releaseConsumers() {
+ // do not leave a consumer waiting when a test fails
+ release.countDown();
+ }
+
+ @Override
+ protected void doPostSetup() {
+ // the exchanges of the two consumers (both are from the disruptor
endpoint)
+ consumersDone = new
NotifyBuilder(context).from("disruptor:fanout*").whenDone(2).create();
+ }
+
+ @Test
+ void testFailureThenSuccess() throws Exception {
+ MockEndpoint slow = getMockEndpoint("mock:slow");
+ slow.expectedBodiesReceived("fail-first");
+
+ template.sendBody("direct:start", "fail-first");
+
+ // the consumer that fails completes the exchange, while the other
consumer is still routing it
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() ->
!events.isEmpty());
+ assertEquals(List.of("failure:fail-first"), events);
+
+ release.countDown();
+ MockEndpoint.assertIsSatisfied(context);
+ // the other consumer is done as well, which must not run the on
completions a second time
+ assertNoMoreEvents(List.of("failure:fail-first"));
+ }
+
+ @Test
+ void testBothFail() throws Exception {
+ MockEndpoint slow = getMockEndpoint("mock:slow");
+ slow.expectedBodiesReceived("both-fail");
+
+ template.sendBody("direct:start", "both-fail");
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() ->
!events.isEmpty());
+ release.countDown();
+ MockEndpoint.assertIsSatisfied(context);
+ assertNoMoreEvents(List.of("failure:both-fail"));
+ }
+
+ @Test
+ void testSuccessThenFailure() throws Exception {
+ MockEndpoint slow = getMockEndpoint("mock:slow");
+ slow.expectedBodiesReceived("success-first");
+
+ NotifyBuilder successDone = new
NotifyBuilder(context).from("disruptor:fanout*").whenDone(1).create();
+ template.sendBody("direct:start", "success-first");
+
+ // the consumer that fails waits until the other consumer is done
+ assertTrue(successDone.matches(10, TimeUnit.SECONDS), "The consumer
that succeeds should be done");
+ slow.assertIsSatisfied();
+ assertTrue(events.isEmpty(), "Events: " + events);
+ release.countDown();
+ assertNoMoreEvents(List.of("failure:success-first"));
+ }
+
+ @Test
+ void testAllSucceed() throws Exception {
+ MockEndpoint slow = getMockEndpoint("mock:slow");
+ slow.expectedBodiesReceived("success");
+
+ template.sendBody("direct:start", "success");
+ release.countDown();
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertNoMoreEvents(List.of("complete:success"));
+ }
+
+ private void assertNoMoreEvents(List<String> expected) {
+ // the consumers call back from the on completions of their own
exchanges, which run before these are done
+ assertTrue(consumersDone.matches(10, TimeUnit.SECONDS), "Both
consumers should be done");
+ assertEquals(expected, events);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .process(e ->
e.getExchangeExtension().addOnCompletion(new SynchronizationAdapter() {
+ @Override
+ public void onComplete(Exchange exchange) {
+ events.add("complete:" +
exchange.getMessage().getBody(String.class));
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ events.add("failure:" +
exchange.getMessage().getBody(String.class));
+ }
+ }))
+ .to("disruptor:fanout?multipleConsumers=true");
+
+ // fails at once, or after the other consumer is done
+
from("disruptor:fanout?multipleConsumers=true").errorHandler(noErrorHandler())
+ .choice()
+ .when(body().isEqualTo("success-first")).process(e ->
release.await(20, TimeUnit.SECONDS))
+ .end()
+ .filter(body().isNotEqualTo("success"))
+ .throwException(new
IllegalArgumentException("Forced"));
+
+ // done when released, and fails too for both-fail
+
from("disruptor:fanout?multipleConsumers=true").errorHandler(noErrorHandler())
+ .choice()
+ .when(body().isNotEqualTo("success-first")).process(e
-> release.await(20, TimeUnit.SECONDS))
+ .end()
+ .to("mock:slow")
+ .filter(body().isEqualTo("both-fail"))
+ .throwException(new
IllegalArgumentException("Forced"));
+ }
+ };
+ }
+}
diff --git
a/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/MultipleConsumerSynchronizedExchangeTest.java
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/MultipleConsumerSynchronizedExchangeTest.java
new file mode 100644
index 000000000000..e8226495c483
--- /dev/null
+++
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/MultipleConsumerSynchronizedExchangeTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.component.disruptor;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The consumers of a multipleConsumers endpoint finish at the same time, one
with a failure: the exchange is completed
+ * once, with the failure. On main the consumer that failed was held after it
was counted, so the other consumer was
+ * counted last and completed the exchange as a success first.
+ */
+public class MultipleConsumerSynchronizedExchangeTest {
+
+ private final CamelContext context = new DefaultCamelContext();
+ private final List<String> events = new CopyOnWriteArrayList<>();
+
+ @AfterEach
+ void stopContext() throws Exception {
+ context.close();
+ }
+
+ @Test
+ void testConsumersFinishTogetherOneFails() throws Exception {
+ Exchange published = new DefaultExchange(context);
+ published.getMessage().setBody("Hello");
+ published.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onComplete(Exchange exchange) {
+ events.add("complete");
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ events.add("failure");
+ }
+ });
+ MultipleConsumerSynchronizedExchange synchronizedExchange = new
MultipleConsumerSynchronizedExchange(published, 2);
+
+ Exchange failedResult = new DefaultExchange(context);
+ failedResult.setException(new IllegalArgumentException("Forced"));
+ // the failing consumer is held when consumed() looks at its exception
for the first time, while the other
+ // consumer, which succeeded, is done
+ CountDownLatch held = new CountDownLatch(1);
+ CountDownLatch go = new CountDownLatch(1);
+ AtomicBoolean first = new AtomicBoolean(true);
+ Exchange heldResult = (Exchange)
Proxy.newProxyInstance(getClass().getClassLoader(), new Class<?>[] {
Exchange.class },
+ (proxy, method, args) -> {
+ if ("getException".equals(method.getName()) &&
method.getParameterCount() == 0
+ && first.getAndSet(false)) {
+ held.countDown();
+ go.await(10, TimeUnit.SECONDS);
+ }
+ try {
+ return method.invoke(failedResult, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+
+ Thread consumer = new Thread(() ->
synchronizedExchange.consumed(heldResult), "failing-consumer");
+ consumer.start();
+ assertTrue(held.await(10, TimeUnit.SECONDS));
+ synchronizedExchange.consumed(new DefaultExchange(context));
+ go.countDown();
+ consumer.join(10000);
+ assertFalse(consumer.isAlive());
+
+ assertEquals(List.of("failure"), events);
+ assertTrue(published.isFailed());
+ }
+}
