This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 77ace31bbca7 [backport camel-4.22.x] CAMEL-24591: camel-seda -
Re-register released queue reference on route restart
77ace31bbca7 is described below
commit 77ace31bbca79bf7b0a856e5d1dc93f52605bb85
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Sep 2 21:18:45 2026 +0200
[backport camel-4.22.x] CAMEL-24591: camel-seda - Re-register released
queue reference on route restart
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
.../camel/component/seda/QueueReference.java | 9 ++++
.../apache/camel/component/seda/SedaEndpoint.java | 27 +++++++----
.../seda/SedaConsumerRouteRestartTest.java | 55 ++++++++++++++++++++++
.../seda/SedaProducerRouteRestartTest.java | 51 ++++++++++++++++++++
.../seda/SedaSharedQueueEndpointRestartTest.java | 55 ++++++++++++++++++++++
5 files changed, 187 insertions(+), 10 deletions(-)
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 28e97fff894a..7b6c42abde85 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
@@ -69,6 +69,15 @@ public final class QueueReference {
}
}
+ boolean isReferenced(SedaEndpoint endpoint) {
+ lock.lock();
+ try {
+ return endpoints.contains(endpoint);
+ } 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 2dc89a4023cb..4ac74dcd5a43 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
@@ -227,14 +227,15 @@ public class SedaEndpoint extends DefaultEndpoint
implements AsyncEndpoint, Brow
public BlockingQueue<Exchange> getQueue() {
lock.lock();
try {
- if (queue == null) {
+ if (queue == null || (getComponent() != null && (ref == null ||
!ref.isReferenced(this)))) {
// prefer to lookup queue from component, so if this endpoint
is re-created or re-started
// then the existing queue from the component can be used, so
new producers and consumers
- // can use the already existing queue referenced from the
component
+ // can use the already existing queue referenced from the
component; a released or
+ // de-listed reference must not be reused as the component no
longer tracks it
if (getComponent() != null) {
// use null to indicate default size (= use what the
existing queue has been configured with)
Integer size = (getSize() == Integer.MAX_VALUE ||
getSize() == SedaConstants.QUEUE_SIZE) ? null : getSize();
- QueueReference ref = getComponent().getOrCreateQueue(this,
size, isMultipleConsumers(), queueFactory);
+ ref = getComponent().getOrCreateQueue(this, size,
isMultipleConsumers(), queueFactory);
queue = ref.getQueue();
String key = getComponent().getQueueKey(getEndpointUri());
LOG.debug("Endpoint {} is using shared queue: {} with
size: {}", this, key,
@@ -605,12 +606,7 @@ public class SedaEndpoint extends DefaultEndpoint
implements AsyncEndpoint, Brow
void onStarted(SedaProducer producer) {
producers.add(producer);
- if (getComponent() != null && (ref == null || queue == null)) {
- // re-register queue reference when producer restarts after queue
was released on stop
- Integer size = (getSize() == Integer.MAX_VALUE || getSize() ==
SedaConstants.QUEUE_SIZE) ? null : getSize();
- ref = getComponent().getOrCreateQueue(this, size,
isMultipleConsumers(), queueFactory);
- queue = ref.getQueue();
- }
+ registerQueueIfStale();
}
void onStopped(SedaProducer producer) {
@@ -623,6 +619,7 @@ public class SedaEndpoint extends DefaultEndpoint
implements AsyncEndpoint, Brow
void onStarted(SedaConsumer consumer) throws Exception {
consumers.add(consumer);
+ registerQueueIfStale();
if (isMultipleConsumers()) {
updateMulticastProcessor();
}
@@ -635,6 +632,17 @@ public class SedaEndpoint extends DefaultEndpoint
implements AsyncEndpoint, Brow
}
}
+ private void registerQueueIfStale() {
+ if (getComponent() != null && (ref == null || queue == null ||
!ref.isReferenced(this))) {
+ // re-register when a producer or consumer restarts after the
queue was released on stop, or
+ // when this endpoint was dropped from a reference still shared
with other endpoints; the
+ // stale ref/queue fields may be non-null while no longer
registered with the component
+ Integer size = (getSize() == Integer.MAX_VALUE || getSize() ==
SedaConstants.QUEUE_SIZE) ? null : getSize();
+ ref = getComponent().getOrCreateQueue(this, size,
isMultipleConsumers(), queueFactory);
+ queue = ref.getQueue();
+ }
+ }
+
public boolean hasConsumers() {
return !this.consumers.isEmpty();
}
@@ -709,5 +717,4 @@ public class SedaEndpoint extends DefaultEndpoint
implements AsyncEndpoint, Brow
queue = null;
ref = null;
}
-
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaConsumerRouteRestartTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaConsumerRouteRestartTest.java
new file mode 100644
index 000000000000..8410b184e255
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaConsumerRouteRestartTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+class SedaConsumerRouteRestartTest extends ContextTestSupport {
+
+ @Test
+ void testConsumerFirstRestartAfterQueueRelease() throws Exception {
+ // stopping the consumer route first and the producer route last
releases the shared queue
+ // reference; restarting the consumer route first must not leave its
pollers on the orphaned
+ // queue while a later restarted producer registers a fresh one
+ context.getRouteController().stopRoute("consumer");
+ context.getRouteController().stopRoute("producer");
+
+ context.getRouteController().startRoute("consumer");
+ context.getRouteController().startRoute("producer");
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("after-restart");
+
+ template.sendBody("direct:start", "after-restart");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("producer").to("seda:bar");
+ from("seda:bar").routeId("consumer").to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaProducerRouteRestartTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaProducerRouteRestartTest.java
new file mode 100644
index 000000000000..cd9a2bf9ffff
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaProducerRouteRestartTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class SedaProducerRouteRestartTest extends ContextTestSupport {
+
+ @Test
+ void testSendAfterProducerRouteRestart() throws Exception {
+ // no call may touch the endpoint's queue reference between stop and
restart: a stale
+ // non-null reference left over from the stop is exactly what this
test exercises
+ context.getRouteController().stopRoute("producer");
+ context.getRouteController().startRoute("producer");
+
+ assertDoesNotThrow(() -> template.sendBody("direct:start",
"after-restart"),
+ "send after producer route restart should be delivered to the
queue, not fail");
+
+ SedaEndpoint bar = getMandatoryEndpoint("seda:bar",
SedaEndpoint.class);
+ assertEquals(1, bar.getCurrentQueueSize());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("producer").to("seda:bar");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSharedQueueEndpointRestartTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSharedQueueEndpointRestartTest.java
new file mode 100644
index 000000000000..d354e0171f0b
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSharedQueueEndpointRestartTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class SedaSharedQueueEndpointRestartTest extends ContextTestSupport {
+
+ @Test
+ void testRestartedEndpointKeepsSharedQueueAlive() throws Exception {
+ // seda:bar and seda:bar?blockWhenFull=true are distinct endpoint
instances sharing one queue
+ // key; a restarted endpoint must be re-added to the shared queue
reference, otherwise stopping
+ // the sibling endpoint later removes the queue while this endpoint is
still active
+ context.getRouteController().stopRoute("a");
+ context.getRouteController().startRoute("a");
+
+ context.getRouteController().stopRoute("b");
+
+ assertDoesNotThrow(() -> template.sendBody("direct:a", "kept-alive"),
+ "send after the sibling endpoint stopped should still find the
shared queue");
+
+ SedaEndpoint bar = getMandatoryEndpoint("seda:bar",
SedaEndpoint.class);
+ assertEquals(1, bar.getCurrentQueueSize());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:a").routeId("a").to("seda:bar");
+
from("direct:b").routeId("b").to("seda:bar?blockWhenFull=true");
+ }
+ };
+ }
+}
