This is an automated email from the ASF dual-hosted git repository.
apupier 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 dca2a14b07c4 CAMEL-26680: camel-disruptor - multipleConsumers: each
consumer copies a snapshot of the exchange
dca2a14b07c4 is described below
commit dca2a14b07c400f91e57e463d5ed7cdba58f0708
Author: Aurélien Pupier <[email protected]>
AuthorDate: Fri Nov 6 10:29:23 2026 +0100
CAMEL-26680: camel-disruptor - multipleConsumers: each consumer copies a
snapshot of the exchange
With multipleConsumers=true each consumer copied the shared original
exchange when it got to the event. The first consumer that failed copied its
result (the exception, body, headers and properties) into that exchange, so a
consumer that came later routed a copy that already had the exception and did
nothing with it, and a consumer that copied at the same moment read the maps
while they were written. The message was silently dropped for that consumer,
which also made DisruptorMultiple [...]
MultipleConsumerSynchronizedExchange now takes a snapshot of the exchange
when it is published, and the consumers copy that one. The test no longer needs
a barrier between the consumers, and a new test holds a consumer until the
other one has failed the exchange.
Co-authored-by: IBM Bob 2.2.1
Signed-off-by: Aurélien Pupier <[email protected]>
---
.../disruptor/AbstractSynchronizedExchange.java | 5 ++
.../component/disruptor/DisruptorConsumer.java | 3 +-
.../MultipleConsumerSynchronizedExchange.java | 11 +++
.../component/disruptor/SynchronizedExchange.java | 7 ++
...DisruptorMultipleConsumersLateConsumerTest.java | 87 ++++++++++++++++++++++
5 files changed, 112 insertions(+), 1 deletion(-)
diff --git
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/AbstractSynchronizedExchange.java
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/AbstractSynchronizedExchange.java
index ae138988a68a..ef100959a53c 100644
---
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/AbstractSynchronizedExchange.java
+++
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/AbstractSynchronizedExchange.java
@@ -36,6 +36,11 @@ public abstract class AbstractSynchronizedExchange
implements SynchronizedExchan
return exchange;
}
+ @Override
+ public Exchange getExchangeToCopy() {
+ return exchange;
+ }
+
@Override
public Exchange cancelAndGetOriginalExchange() {
if (synchronizations != null) {
diff --git
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java
index f20145996c07..639994443bbd 100644
---
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java
+++
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java
@@ -135,6 +135,7 @@ public class DisruptorConsumer extends ServiceSupport
implements Consumer, Suspe
private Exchange prepareExchange(final Exchange exchange) {
// send a new copied exchange with new camel context
// don't copy handovers as they are handled by the Disruptor Event
Handlers
+ // (with several consumers the exchange is a snapshot, as another
consumer may fail and change the original)
final Exchange newExchange =
ExchangeHelper.copyExchangeWithProperties(exchange, endpoint.getCamelContext());
// the flag is only for this consumer, and must not be routed (or
copied back to the caller)
newExchange.removeProperty(DisruptorEndpoint.DISRUPTOR_IGNORE_EXCHANGE);
@@ -154,7 +155,7 @@ public class DisruptorConsumer extends ServiceSupport
implements Consumer, Suspe
}
// send a new copied exchange with new camel context
- final Exchange result = prepareExchange(exchange);
+ final Exchange result =
prepareExchange(synchronizedExchange.getExchangeToCopy());
// We need to be notified when the exchange processing is complete
to synchronize the original exchange
// This is however, the last part of the processing of this
exchange and as such can't be done
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 b2f2df6abd35..cc82c88247ad 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
@@ -30,6 +30,10 @@ public class MultipleConsumerSynchronizedExchange extends
AbstractSynchronizedEx
private final int expectedConsumers;
+ // the consumers copy this one: the first consumer that fails copies its
result into the exchange, which the other
+ // consumers must neither see nor read while it is written
+ private final Exchange exchangeToCopy;
+
private final AtomicInteger processedConsumers = new AtomicInteger();
private final AtomicBoolean resultHandled = new AtomicBoolean();
@@ -40,6 +44,13 @@ public class MultipleConsumerSynchronizedExchange extends
AbstractSynchronizedEx
super(exchange);
this.expectedConsumers = expectedConsumers;
processedConsumers.set(0);
+ // taken when the exchange is published, before any consumer can fail
+ this.exchangeToCopy =
ExchangeHelper.copyExchangeWithProperties(exchange, exchange.getContext());
+ }
+
+ @Override
+ public Exchange getExchangeToCopy() {
+ return exchangeToCopy;
}
@Override
diff --git
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/SynchronizedExchange.java
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/SynchronizedExchange.java
index ef51243ce9b0..d82344706f4c 100644
---
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/SynchronizedExchange.java
+++
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/SynchronizedExchange.java
@@ -25,6 +25,13 @@ import org.apache.camel.Exchange;
public interface SynchronizedExchange {
Exchange getExchange();
+ /**
+ * The exchange that a consumer copies to route. With several consumers
this is a snapshot taken when the exchange
+ * is published, as the first consumer that fails copies its result into
{@link #getExchange()}, which the other
+ * consumers must not see (the exception) or read while it is written.
+ */
+ Exchange getExchangeToCopy();
+
void consumed(Exchange result);
Exchange cancelAndGetOriginalExchange();
diff --git
a/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorMultipleConsumersLateConsumerTest.java
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorMultipleConsumersLateConsumerTest.java
new file mode 100644
index 000000000000..0162391adb04
--- /dev/null
+++
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorMultipleConsumersLateConsumerTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With multipleConsumers every consumer routes its own copy of the exchange,
also a consumer that gets to the exchange
+ * after another consumer has failed it.
+ */
+public class DisruptorMultipleConsumersLateConsumerTest extends
CamelTestSupport {
+
+ private final CountDownLatch blockerStarted = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @AfterEach
+ void releaseConsumers() {
+ // do not leave a consumer waiting when a test fails
+ release.countDown();
+ }
+
+ @Test
+ void testLateConsumerRoutesTheExchangeThatAnotherConsumerFailed() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:late");
+ mock.expectedBodiesReceived("blocker", "fail");
+
+ NotifyBuilder failingDone = new
NotifyBuilder(context).from("disruptor:late*").whenDone(2).create();
+
+ // the late consumer is busy with this one, so it gets to the next
exchange only when it is released
+ template.sendBody("direct:start", "blocker");
+ assertTrue(blockerStarted.await(10, TimeUnit.SECONDS), "The late
consumer should be busy");
+
+ // the other consumer is done with both exchanges, and has failed the
second one
+ template.sendBody("direct:start", "fail");
+ assertTrue(failingDone.matches(10, TimeUnit.SECONDS), "The failing
consumer should be done");
+
+ release.countDown();
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").to("disruptor:late?multipleConsumers=true");
+
+
from("disruptor:late?multipleConsumers=true").errorHandler(noErrorHandler())
+ .filter(body().isEqualTo("fail"))
+ .throwException(new
IllegalArgumentException("Forced"));
+
+
from("disruptor:late?multipleConsumers=true").errorHandler(noErrorHandler())
+ .filter(body().isEqualTo("blocker"))
+ .process(e -> {
+ blockerStarted.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .end()
+ .to("mock:late");
+ }
+ };
+ }
+}
