CAMEL-24591: camel-seda - Re-register released queue referenced

components/camel-seda/src/main/java/org/apache/camel/component/seda/QueueReference.java

@@ -69,6 +69,15 @@ void removeReference(SedaEndpoint endpoint) {
         }
     }
 
+    boolean isReferenced(SedaEndpoint endpoint) {
+        lock.lock();
+        try {
+            return endpoints.contains(endpoint);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     /**
      * Gets the reference counter
      */

components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java

@@ -227,14 +227,15 @@ public PollingConsumer createPollingConsumer() throws Exception {
     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 Set<SedaProducer> getProducers() {
 
     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 @@ void onStopped(SedaProducer producer) {
 
     void onStarted(SedaConsumer consumer) throws Exception {
         consumers.add(consumer);
+        registerQueueIfStale();
         if (isMultipleConsumers()) {
             updateMulticastProcessor();
         }
@@ -635,6 +632,17 @@ void onStopped(SedaConsumer consumer) throws Exception {
         }
     }
 
+    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 @@ protected void doShutdown() throws Exception {
         queue = null;
         ref = null;
     }
-
 }

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");
+            }
+        };
+    }
+}

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");
+            }
+        };
+    }
+}

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");
+            }
+        };
+    }
+}