You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/03/05 11:10:42 UTC
(camel) branch seda created (now b2840cc04123)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch seda
in repository https://gitbox.apache.org/repos/asf/camel.git
at b2840cc04123 CAMEL-23129: Fix seda virtual threads hang. Add also shutdown timeout for non vitual thread for just in case.
This branch includes the following new commits:
new b2840cc04123 CAMEL-23129: Fix seda virtual threads hang. Add also shutdown timeout for non vitual thread for just in case.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-23129: Fix seda virtual threads hang. Add also shutdown timeout for non vitual thread for just in case.
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch seda
in repository https://gitbox.apache.org/repos/asf/camel.git
commit b2840cc04123a7191a4f6790c035361f03c2f810
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Mar 5 12:10:25 2026 +0100
CAMEL-23129: Fix seda virtual threads hang. Add also shutdown timeout for non vitual thread for just in case.
---
.../apache/camel/component/seda/SedaConsumer.java | 23 +++++++++++++++++-----
.../seda/ThreadPerTaskSedaConsumerTest.java | 5 +++--
.../ManagedVirtualThreadExecutorTest.java | 5 -----
3 files changed, 21 insertions(+), 12 deletions(-)
diff --git a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
index 845fd9045513..dcc6b7296074 100644
--- a/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
+++ b/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
@@ -48,6 +48,8 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA
private static final Logger LOG = LoggerFactory.getLogger(SedaConsumer.class);
+ private static final long SHUTDOWN_TIMEOUT = 30000L;
+
private final AtomicInteger taskCount = new AtomicInteger();
private volatile CountDownLatch latch;
private volatile boolean shutdownPending;
@@ -112,7 +114,11 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA
// wait for all threads to end
try {
- latch.await();
+ boolean zero = latch.await(SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);
+ if (!zero) {
+ LOG.warn("Timeout {}ms preparing to shutdown, waiting for {} consumer threads to complete.",
+ SHUTDOWN_TIMEOUT, latch.getCount());
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
@@ -141,8 +147,10 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA
doRun();
} finally {
taskCount.decrementAndGet();
- latch.countDown();
- LOG.debug("Ending this polling consumer thread, there are still {} consumer threads left.", latch.getCount());
+ if (latch != null) {
+ latch.countDown();
+ LOG.debug("Ending this polling consumer thread, there are still {} consumer threads left.", latch.getCount());
+ }
}
}
@@ -348,7 +356,9 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA
@Override
protected void doStart() throws Exception {
super.doStart();
- latch = new CountDownLatch(getEndpoint().getConcurrentConsumers());
+ if (getEndpoint().getConcurrentConsumers() > 0) {
+ latch = new CountDownLatch(getEndpoint().getConcurrentConsumers());
+ }
shutdownPending = false;
forceShutdown = false;
@@ -421,7 +431,10 @@ public class SedaConsumer extends DefaultConsumer implements Runnable, ShutdownA
// submit needed number of tasks
int tasks = poolSize - taskCount.get();
- LOG.debug("Creating {} consumer tasks with poll timeout {} ms.", tasks, pollTimeout);
+ if (tasks <= 0) {
+ tasks = 1;
+ }
+ LOG.debug("Creating {} queue pooler with poll timeout {} ms.", tasks, pollTimeout);
for (int i = 0; i < tasks; i++) {
executor.execute(this);
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java b/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java
index b0f6e8c3890d..b3cf9e5af948 100644
--- a/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/component/seda/ThreadPerTaskSedaConsumerTest.java
@@ -19,13 +19,11 @@ 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.Disabled;
import org.junit.jupiter.api.Test;
/**
* Test for the virtualThreadPerTask mode of SEDA consumer
*/
-@Disabled
public class ThreadPerTaskSedaConsumerTest extends ContextTestSupport {
@Test
@@ -71,12 +69,15 @@ public class ThreadPerTaskSedaConsumerTest extends ContextTestSupport {
@Override
public void configure() {
from("seda:test?virtualThreadPerTask=true")
+ .to("log:result")
.to("mock:result");
from("seda:limited?virtualThreadPerTask=true&concurrentConsumers=2")
+ .to("log:limited")
.to("mock:limited");
from("seda:throughput?virtualThreadPerTask=true")
+ .to("log:throughput")
.to("mock:throughput");
}
};
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/ManagedVirtualThreadExecutorTest.java b/core/camel-management/src/test/java/org/apache/camel/management/ManagedVirtualThreadExecutorTest.java
index e2004883f302..e6e4d500689d 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/ManagedVirtualThreadExecutorTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/ManagedVirtualThreadExecutorTest.java
@@ -25,18 +25,13 @@ import javax.management.ObjectName;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.spi.LifecycleStrategy;
-import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.EnabledForJreRange;
-import org.junit.jupiter.api.condition.JRE;
import static org.apache.camel.management.DefaultManagementObjectNameStrategy.TYPE_THREAD_POOL;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@Disabled
-@EnabledForJreRange(min = JRE.JAVA_21)
public class ManagedVirtualThreadExecutorTest extends ManagementTestSupport {
private ExecutorService vte;