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
commit d876d4b6e887af59b0acef08ab38d2aa3df9add1 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 18:04:05 2026 +0530 CAMEL-24942: camel-seda - multicast to all consumers of a queue when multipleConsumers endpoints use different options Cause: SEDA queues are shared by queue name (the uri query is dropped from the queue key), but each distinct endpoint uri is its own SedaEndpoint with its own consumer set and multicast processor. A consumer thread multicasts a polled message only to the consumers of its own endpoint. Effect: with multipleConsumers=true, subscribers whose uris differ in any other option (e.g. concurrentConsumers or pollTimeout) silently compete for the messages instead of each receiving a copy, although the option is documented as publish/subscribe and only has to be set on every consumer. Fix: the multicast processor of a multipleConsumers endpoint now includes the active consumers of every multipleConsumers endpoint that shares the same QueueReference, and a consumer that starts, stops, suspends or resumes updates the multicast processors of the other endpoints of the queue that have consumers. Consumers on the same endpoint behave as before. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../apache/camel/catalog/docs/seda-component.adoc | 5 + .../camel-seda/src/main/docs/seda-component.adoc | 5 + .../camel/component/seda/QueueReference.java | 13 +++ .../apache/camel/component/seda/SedaEndpoint.java | 49 +++++++-- .../SedaMultipleConsumersDifferentOptionsTest.java | 117 +++++++++++++++++++++ 5 files changed, 183 insertions(+), 6 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/seda-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/seda-component.adoc index 79d476a3cb51..b9b9befc43ec 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/seda-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/seda-component.adoc @@ -340,6 +340,11 @@ Since we have specified `multipleConsumers=true` on the seda `foo` endpoint we can have those two consumers receive their own copy of the message as a kind of _publish/subscribe_ style messaging. +The consumers share the queue by its name, so they can use different consumer options, +such as `seda:foo?multipleConsumers=true&concurrentConsumers=5`, and still each receive +their own copy of the message. The `multipleConsumers=true` option must be specified +on every consumer. + === Extracting queue information. If needed, information such as queue size, etc. can be obtained without diff --git a/components/camel-seda/src/main/docs/seda-component.adoc b/components/camel-seda/src/main/docs/seda-component.adoc index 79d476a3cb51..b9b9befc43ec 100644 --- a/components/camel-seda/src/main/docs/seda-component.adoc +++ b/components/camel-seda/src/main/docs/seda-component.adoc @@ -340,6 +340,11 @@ Since we have specified `multipleConsumers=true` on the seda `foo` endpoint we can have those two consumers receive their own copy of the message as a kind of _publish/subscribe_ style messaging. +The consumers share the queue by its name, so they can use different consumer options, +such as `seda:foo?multipleConsumers=true&concurrentConsumers=5`, and still each receive +their own copy of the message. The `multipleConsumers=true` option must be specified +on every consumer. + === Extracting queue information. If needed, information such as queue size, etc. can be obtained without diff --git a/components/camel-seda/src/main/java/org/apache/camel/component/seda/QueueReference.java b/components/camel-seda/src/main/java/org/apache/camel/component/seda/QueueReference.java index 7b6c42abde85..30d9be3579d5 100644 --- a/components/camel-seda/src/main/java/org/apache/camel/component/seda/QueueReference.java +++ b/components/camel-seda/src/main/java/org/apache/camel/component/seda/QueueReference.java @@ -16,6 +16,7 @@ */ package org.apache.camel.component.seda; +import java.util.ArrayList; import java.util.LinkedList; import java.util.List; import java.util.concurrent.BlockingQueue; @@ -78,6 +79,18 @@ public final class QueueReference { } } + /** + * Gets a snapshot of the endpoints sharing this queue reference. + */ + List<SedaEndpoint> getEndpoints() { + lock.lock(); + try { + return new ArrayList<>(endpoints); + } finally { + lock.unlock(); + } + } + /** * Gets the reference counter */ diff --git a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java index 57e0eafbd376..6ce9d7ec6510 100644 --- a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java +++ b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java @@ -18,6 +18,7 @@ package org.apache.camel.component.seda; import java.util.ArrayList; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Set; import java.util.concurrent.BlockingQueue; @@ -317,16 +318,18 @@ public class SedaEndpoint extends DefaultEndpoint implements AsyncEndpoint, Brow consumerMulticastProcessor = null; } - int size = getConsumers().size(); - if (size >= 1) { + // the consumer threads of this endpoint multicast to every consumer of the (shared) queue, + // which includes consumers on other endpoints for the same queue name that use different uri options + if (!getConsumers().isEmpty()) { if (multicastExecutor == null) { // create multicast executor as we need it when we have more than 1 processor multicastExecutor = getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, URISupport.sanitizeUri(getEndpointUri()) + "(multicast)"); } // create list of consumers to multicast to - List<Processor> processors = new ArrayList<>(size); - for (SedaConsumer consumer : getConsumers()) { + Set<SedaConsumer> queueConsumers = getQueueConsumers(); + List<Processor> processors = new ArrayList<>(queueConsumers.size()); + for (SedaConsumer consumer : queueConsumers) { processors.add(consumer.getProcessor()); } // create multicast processor @@ -341,6 +344,40 @@ public class SedaEndpoint extends DefaultEndpoint implements AsyncEndpoint, Brow } } + /** + * Gets the active consumers of this endpoint and of the other multiple consumers endpoints that share the same + * queue, as they may be using the same queue name with different uri options. + */ + private Set<SedaConsumer> getQueueConsumers() { + Set<SedaConsumer> answer = new LinkedHashSet<>(getConsumers()); + QueueReference queueReference = ref; + if (queueReference != null) { + for (SedaEndpoint endpoint : queueReference.getEndpoints()) { + if (endpoint != this && endpoint.isMultipleConsumers()) { + answer.addAll(endpoint.getConsumers()); + } + } + } + return answer; + } + + /** + * Updates the multicast processor of this endpoint, and of the other multiple consumers endpoints with active + * consumers that share the same queue, so they all multicast to the current set of consumers of the queue. + */ + private void updateMulticastProcessors() throws Exception { + updateMulticastProcessor(); + QueueReference queueReference = ref; + if (queueReference != null) { + for (SedaEndpoint endpoint : queueReference.getEndpoints()) { + if (endpoint != this && endpoint.isMultipleConsumers() && endpoint.hasConsumers()) { + // do not hold the lock of this endpoint while updating the other endpoint + endpoint.updateMulticastProcessor(); + } + } + } + } + void setName(String name) { this.name = name; } @@ -631,14 +668,14 @@ public class SedaEndpoint extends DefaultEndpoint implements AsyncEndpoint, Brow consumers.add(consumer); registerQueueIfStale(); if (isMultipleConsumers()) { - updateMulticastProcessor(); + updateMulticastProcessors(); } } void onStopped(SedaConsumer consumer) throws Exception { consumers.remove(consumer); if (isMultipleConsumers()) { - updateMulticastProcessor(); + updateMulticastProcessors(); } } diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaMultipleConsumersDifferentOptionsTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaMultipleConsumersDifferentOptionsTest.java new file mode 100644 index 000000000000..8bf577abd39a --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaMultipleConsumersDifferentOptionsTest.java @@ -0,0 +1,117 @@ +/* + * 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.seda; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; + +/** + * With multipleConsumers=true every consumer of the queue must receive a copy of each message, also when the consumer + * endpoints use the same queue name with different uri options. + */ +public class SedaMultipleConsumersDifferentOptionsTest extends ContextTestSupport { + + private static final int MESSAGES = 20; + + @Test + public void testEachConsumerReceivesEveryMessage() throws Exception { + String[] bodies = bodies("Hello"); + getMockEndpoint("mock:a").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + getMockEndpoint("mock:b").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + getMockEndpoint("mock:c").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + + for (String body : bodies) { + template.sendBody("seda:news", body); + } + + assertMockEndpointsSatisfied(); + } + + @Test + public void testStoppedConsumer() throws Exception { + context.getRouteController().stopRoute("b"); + + String[] bodies = bodies("Bye"); + getMockEndpoint("mock:a").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + getMockEndpoint("mock:b").expectedMessageCount(0); + getMockEndpoint("mock:c").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + + for (String body : bodies) { + template.sendBody("seda:news", body); + } + + assertMockEndpointsSatisfied(); + + // and when started again it receives the messages sent from now on as well + context.getRouteController().startRoute("b"); + resetMocks(); + + bodies = bodies("Again"); + getMockEndpoint("mock:a").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + getMockEndpoint("mock:b").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + getMockEndpoint("mock:c").expectedBodiesReceivedInAnyOrder((Object[]) bodies); + + for (String body : bodies) { + template.sendBody("seda:news", body); + } + + assertMockEndpointsSatisfied(); + } + + @Test + public void testConsumerAddedWithOtherOptions() throws Exception { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("seda:news?multipleConsumers=true&concurrentConsumers=3").routeId("d").to("mock:d"); + } + }); + + String[] bodies = bodies("Hi"); + for (String name : new String[] { "mock:a", "mock:b", "mock:c", "mock:d" }) { + getMockEndpoint(name).expectedBodiesReceivedInAnyOrder((Object[]) bodies); + } + + for (String body : bodies) { + template.sendBody("seda:news", body); + } + + MockEndpoint.assertIsSatisfied(context); + } + + private static String[] bodies(String prefix) { + String[] bodies = new String[MESSAGES]; + for (int i = 0; i < MESSAGES; i++) { + bodies[i] = prefix + " " + i; + } + return bodies; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("seda:news?multipleConsumers=true").routeId("a").to("mock:a"); + from("seda:news?multipleConsumers=true&concurrentConsumers=2").routeId("b").to("mock:b"); + from("seda:news?multipleConsumers=true&pollTimeout=200").routeId("c").to("mock:c"); + } + }; + } +}
