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 6c343ef27820 CAMEL-26605: camel-seda - receive() without a timeout
returns null once the polling consumer is stopped (#29375)
6c343ef27820 is described below
commit 6c343ef27820df3e557353c91b0675bded2f5007
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Nov 7 13:31:38 2026 +0530
CAMEL-26605: camel-seda - receive() without a timeout returns null once the
polling consumer is stopped (#29375)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/seda/SedaPollingConsumer.java | 27 ++++-
.../seda/SedaPollingConsumerStopTest.java | 124 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++
3 files changed, 160 insertions(+), 1 deletion(-)
diff --git
a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
index 86add56a6561..0c2206511d45 100644
---
a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
+++
b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaPollingConsumer.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.seda;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.camel.Endpoint;
@@ -24,9 +25,16 @@ import org.apache.camel.IsSingleton;
import org.apache.camel.Processor;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.support.PollingConsumerSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class SedaPollingConsumer extends PollingConsumerSupport implements
IsSingleton {
+ private static final Logger LOG =
LoggerFactory.getLogger(SedaPollingConsumer.class);
+
+ // how often receive() checks, while it waits for a message, whether this
consumer was stopped
+ private static final long RUN_ALLOWED_CHECK_INTERVAL = 1000;
+
public SedaPollingConsumer(Endpoint endpoint) {
super(endpoint);
}
@@ -43,8 +51,20 @@ public class SedaPollingConsumer extends
PollingConsumerSupport implements IsSin
@Override
public Exchange receive() {
+ BlockingQueue<Exchange> queue = getEndpoint().getQueue();
try {
- return getEndpoint().getQueue().take();
+ while (true) {
+ Exchange answer = queue.poll(RUN_ALLOWED_CHECK_INTERVAL,
TimeUnit.MILLISECONDS);
+ if (answer != null) {
+ return answer;
+ }
+ // once this consumer is stopped (such as when the
CamelContext, the consumer template or the route of a
+ // pollEnrich stops) stop waiting; a consumer that was never
started keeps waiting, as before
+ if (isStoppedAfterStart()) {
+ LOG.trace("Consumer is stopped, so returning null");
+ return null;
+ }
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw RuntimeCamelException.wrapRuntimeCamelException(e);
@@ -76,6 +96,11 @@ public class SedaPollingConsumer extends
PollingConsumerSupport implements IsSin
// noop
}
+ private boolean isStoppedAfterStart() {
+ // stopping, stopped, shutting down, shut down or failed
+ return status >= STOPPING;
+ }
+
@Override
public boolean isSingleton() {
return true;
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPollingConsumerStopTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPollingConsumerStopTest.java
new file mode 100644
index 000000000000..ebb32a6a455f
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/seda/SedaPollingConsumerStopTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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 java.util.Arrays;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.PollingConsumer;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * A receive() without a timeout on a seda endpoint must return once the
polling consumer is stopped, and must keep
+ * waiting on a polling consumer that was never started.
+ */
+class SedaPollingConsumerStopTest extends ContextTestSupport {
+
+ private final ExecutorService executor =
Executors.newSingleThreadExecutor();
+ private volatile Thread receiver;
+
+ @AfterEach
+ void stopExecutor() {
+ // a receive that still waits (the bug) must not keep the test JVM busy
+ executor.shutdownNow();
+ }
+
+ @Test
+ void testConsumerTemplateReceiveReturnsWhenCamelContextStops() throws
Exception {
+ Future<Exchange> received = receiveInThread(() ->
consumer.receive("seda:idle"));
+
+ context.stop();
+
+ assertNull(received.get(10, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void testConsumerTemplateReceiveReturnsWhenTemplateStops() throws
Exception {
+ Future<Exchange> received = receiveInThread(() ->
consumer.receive("seda:idle"));
+
+ consumer.stop();
+
+ assertNull(received.get(10, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void testPollEnrichWithoutTimeoutReturnsWhenRouteStops() throws Exception {
+ Future<Exchange> sent = receiveInThread(() ->
template.send("direct:start", e -> e.getMessage().setBody("x")));
+
+ // the graceful stop waits for the exchange in pollEnrich, the forced
stop then stops the polling consumer
+ context.getRouteController().stopRoute("poll", 1, TimeUnit.SECONDS);
+
+ // and the waiting receive() returns, so the exchange completes
+ sent.get(10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void testReceiveOnPollingConsumerNotStartedKeepsWaiting() throws Exception
{
+ PollingConsumer pollingConsumer =
context.getEndpoint("seda:idle").createPollingConsumer();
+
+ Future<Exchange> received = receiveInThread(pollingConsumer::receive);
+ // longer than the interval in which receive checks whether it was
stopped
+ await().during(1500, TimeUnit.MILLISECONDS).atMost(5,
TimeUnit.SECONDS).until(() -> !received.isDone());
+
+ template.sendBody("seda:idle", "hello");
+
+ assertEquals("hello", received.get(10,
TimeUnit.SECONDS).getMessage().getBody());
+ }
+
+ private Future<Exchange> receiveInThread(Callable<Exchange> receive) {
+ CompletableFuture<Exchange> answer = new CompletableFuture<>();
+ executor.execute(() -> {
+ receiver = Thread.currentThread();
+ try {
+ answer.complete(receive.call());
+ } catch (Exception e) {
+ answer.completeExceptionally(e);
+ }
+ });
+ // wait until the thread waits for a message in the seda polling
consumer
+ await().atMost(10, TimeUnit.SECONDS).until(() -> receiver != null
+ && Arrays.stream(receiver.getStackTrace())
+ .anyMatch(f ->
f.getClassName().equals(SedaPollingConsumer.class.getName())
+ && f.getMethodName().equals("receive")));
+ return answer;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("poll")
+ .pollEnrich("seda:idle")
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index d66a5266da36..f0a3e7e0ec09 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -678,6 +678,16 @@ and retry the call once, and only when they have an
`apiKey` or a `password` to
component get a new token when the current one expires: when the `authToken`
is set, the component uses only that
token.
+=== camel-seda - receive without a timeout returns when the polling consumer
stops
+
+A `receive()` without a timeout on a seda endpoint, such as
`consumerTemplate.receive("seda:foo")` or a `pollEnrich`
+from seda without a timeout, waited for a message for ever, also after the
CamelContext, the consumer template or the
+route of the `pollEnrich` was stopped. It now returns `null` (no message)
within about a second once the polling
+consumer is stopped, as `EventDrivenPollingConsumer`, used by most other
endpoints, does. This also happens when the
+consumer cache of a consumer template evicts the polling consumer (see
`maximumCacheSize`). A message that is in the
+queue is still returned. A polling consumer that was never started still waits
for a message, as before. A caller that
+loops on `receive()` should stop on a `null` result.
+
=== camel-main - a failure to start is logged
A Camel Main application that fails to start now logs why, at ERROR: `Error
starting Camel: ...` with the stack trace,
