From 68da9ac90cb87849a8bca4cb60b460bc3124020a Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Tue, 20 Oct 2026 07:20:05 +0530 Subject: [PATCH 1/2] CAMEL-25840: camel-hazelcast - the SEDA consumer waits for its threads before it stops the processor HazelcastSedaConsumer.doStop shut its executor down without waiting and then stopped the processor, while its threads could still be waiting in poll(pollTimeout) (1 second by default). A poll takes the item from the queue: an item offered in that time was taken, its exchange failed with RouteStoppingException and, without transacted, the item was lost. doStop now waits for the consumer threads to end their current poll and exchange (a graceful executor shutdown) before it stops the processor, and shuts the error-delay scheduler down after them, since the threads use it: a thread in the delay after a failed poll waits for a task of that scheduler, which a shutdown cancels, so it was left waiting for ever when the scheduler was shut down first. Co-Authored-By: Claude Opus 5.5 --- .../hazelcast/seda/HazelcastSedaConsumer.java | 11 +- .../HazelcastSedaConsumerStopTest.java | 153 ++++++++++++++++++ 2 files changed, 160 insertions(+), 4 deletions(-) create mode 100644 components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java index a444a6c19d420..26e24211b91a6 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java @@ -71,14 +71,17 @@ protected void doStart() throws Exception { @Override protected void doStop() throws Exception { + if (executor != null) { + // wait for the consumer threads to end their current poll (at most pollTimeout) and route what it returned: + // a poll takes the item from the queue, so an item polled after the processor is stopped is lost + endpoint.getCamelContext().getExecutorServiceManager().shutdownGraceful(executor); + executor = null; + } + // after the consumer threads, which use it for the delay after an error if (taskExecutor != null) { endpoint.getCamelContext().getExecutorServiceManager().shutdown(taskExecutor); taskExecutor = null; } - if (executor != null) { - endpoint.getCamelContext().getExecutorServiceManager().shutdown(executor); - executor = null; - } super.doStop(); } diff --git a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java new file mode 100644 index 0000000000000..83eba9b54534f --- /dev/null +++ b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java @@ -0,0 +1,153 @@ +/* + * 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.hazelcast; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import com.hazelcast.collection.IQueue; +import com.hazelcast.config.Config; +import com.hazelcast.core.Hazelcast; +import com.hazelcast.core.HazelcastException; +import com.hazelcast.core.HazelcastInstance; +import org.apache.camel.CamelContext; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +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.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Stopping a route that consumes a Hazelcast SEDA queue. A poll removes the item from the queue, so the consumer must + * not stop its processor while its threads can still take an item, and its threads must end also when they wait in the + * delay after a failed poll. + */ +class HazelcastSedaConsumerStopTest extends CamelTestSupport { + + private HazelcastInstance member; + + @Override + protected CamelContext createCamelContext() throws Exception { + member = Hazelcast.newHazelcastInstance(memberConfig("seda-stop-" + UUID.randomUUID())); + CamelContext context = super.createCamelContext(); + context.getRegistry().bind("hazelcastInstance", member); + // an instance whose queue fails every poll, so the consumer waits in its delay after a failed poll + IQueue failingQueue = mock(IQueue.class); + when(failingQueue.poll(anyLong(), any(TimeUnit.class))).thenThrow(new HazelcastException("poll failed")); + HazelcastInstance failingInstance = mock(HazelcastInstance.class); + when(failingInstance.getQueue("failing")).thenReturn(failingQueue); + context.getRegistry().bind("failingInstance", failingInstance); + return context; + } + + private static Config memberConfig(String clusterName) { + Config config = new Config(); + config.setClusterName(clusterName); + config.getMetricsConfig().setEnabled(false); + config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false); + config.getNetworkConfig().getJoin().getAutoDetectionConfig().setEnabled(false); + config.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(false); + return config; + } + + @AfterEach + void shutdownMember() { + // stop the routes before the member, so the consumers do not poll a member that is shut down + context.stop(); + if (member != null) { + member.shutdown(); + } + } + + @Test + void itemOfferedAfterTheRouteStoppedStaysInTheQueue() throws Exception { + // stop the route while the consumer waits in its poll (the usual state of an idle consumer) + await().atMost(10, TimeUnit.SECONDS).until(() -> Thread.getAllStackTraces().entrySet().stream() + .anyMatch(e -> e.getKey().getName().contains("hazelcast-seda://idle") + && Arrays.stream(e.getValue()).anyMatch(f -> "poll".equals(f.getMethodName())))); + context.getRouteController().stopRoute("idle"); + + member.getQueue("idle").offer("b"); + + // nothing consumes the queue any more: the item must stay there (for the next start or another consumer) + await().during(4, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1, member.getQueue("idle").size(), + "The item offered after the route stopped was taken from the queue")); + assertEquals(0, getMockEndpoint("mock:idle").getReceivedCounter()); + + // and the next start routes it + MockEndpoint mock = getMockEndpoint("mock:idle"); + mock.expectedBodiesReceived("b"); + context.getRouteController().startRoute("idle"); + mock.assertIsSatisfied(); + } + + @Test + void threadInTheDelayAfterAFailedPollEndsWhenTheRouteStops() throws Exception { + // the consumer thread waits in the delay after a failed poll, which runs on the error-delay scheduler + await().atMost(10, TimeUnit.SECONDS).until(() -> !failingConsumerThreads(true).isEmpty()); + + long start = System.nanoTime(); + context.getRouteController().stopRoute("failing"); + long stopMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + + // the scheduler is shut down after the consumer threads ended, so the delay ends and the thread sees the stop + // (shut down first, the delay never ends: the stop waits for the graceful timeout of 10 s, or the thread is left + // waiting for ever) + assertTrue(stopMillis < 8000, "Stopping the route took " + stopMillis + " ms"); + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertTrue(failingConsumerThreads(false).isEmpty(), + "A consumer thread is left: " + failingConsumerThreads(false))); + } + + /** + * The threads of the consumer of the failing route, only those waiting in the error delay if asked. + */ + private static List failingConsumerThreads(boolean inErrorDelay) { + return Thread.getAllStackTraces().entrySet().stream() + .filter(e -> e.getKey().getName().contains("hazelcast-seda://failing")) + .filter(e -> !inErrorDelay || Arrays.stream(e.getValue()) + .anyMatch(f -> f.getClassName().endsWith("BackgroundTask"))) + .map(Map.Entry::getKey) + .toList(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("hazelcast-seda:idle?hazelcastInstance=#hazelcastInstance").routeId("idle") + .to("mock:idle"); + + from("hazelcast-seda:failing?hazelcastInstance=#failingInstance&onErrorDelay=3000").routeId("failing") + .to("mock:failing"); + } + }; + } +} From ee5eb3f0524e7c501a568aa3ee49c8044fd552dc Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:17:29 +0530 Subject: [PATCH 2/2] CAMEL-25840: camel-hazelcast - cap a single poll of the SEDA consumer at one second Per review: the graceful wait in doStop only helped while pollTimeout was below shutdownAwaitTermination; with a larger pollTimeout every stop waited 10 s and then interrupted the poll, which in Hazelcast 5.4 does not end the poll on the member. A single poll now waits Math.min(pollTimeout, 1000) ms (as the queue consumer does since CAMEL-25837); a poll that returns nothing goes round the loop and commits the empty transaction as before. New test longPollTimeoutStillStopsPromptlyAndLeavesTheItemInTheQueue (pollTimeout=60000). Co-Authored-By: Claude Opus 5.5 --- .../hazelcast/seda/HazelcastSedaConsumer.java | 8 ++++-- .../HazelcastSedaConsumerStopTest.java | 27 +++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java index 26e24211b91a6..bbcfabb9b10e9 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java @@ -43,6 +43,9 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(HazelcastSedaConsumer.class); + // the longest a single poll waits (or pollTimeout when it is shorter): the consumer stops by letting its poll end, + // so the thread sees a stop within this time. A poll that returns nothing just goes round the loop again + private static final long MAX_POLL_WAIT = 1000L; private final HazelcastSedaEndpoint endpoint; private final AsyncProcessor processor; @@ -72,7 +75,7 @@ protected void doStart() throws Exception { @Override protected void doStop() throws Exception { if (executor != null) { - // wait for the consumer threads to end their current poll (at most pollTimeout) and route what it returned: + // wait for the consumer threads to end their current poll (at most MAX_POLL_WAIT) and route what it returned: // a poll takes the item from the queue, so an item polled after the processor is stopped is lost endpoint.getCamelContext().getExecutorServiceManager().shutdownGraceful(executor); executor = null; @@ -105,7 +108,8 @@ public void run() { } } - final Object body = queue.poll(endpoint.getConfiguration().getPollTimeout(), TimeUnit.MILLISECONDS); + final Object body = queue.poll(Math.min(endpoint.getConfiguration().getPollTimeout(), MAX_POLL_WAIT), + TimeUnit.MILLISECONDS); if (body != null) { if (body instanceof DefaultExchangeHolder defaultExchangeHolder) { diff --git a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java index 83eba9b54534f..eb319afb4ec97 100644 --- a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java +++ b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaConsumerStopTest.java @@ -107,6 +107,30 @@ void itemOfferedAfterTheRouteStoppedStaysInTheQueue() throws Exception { mock.assertIsSatisfied(); } + @Test + void longPollTimeoutStillStopsPromptlyAndLeavesTheItemInTheQueue() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:long"); + mock.expectedBodiesReceived("a"); + member.getQueue("long").offer("a"); + mock.assertIsSatisfied(); + + // with a pollTimeout of a minute the stop does not wait for the poll (nor for the graceful shutdown timeout, which + // would interrupt it; an interrupt does not end a poll on the member, so the poll would take the next item) + await().atMost(10, TimeUnit.SECONDS).until(() -> Thread.getAllStackTraces().entrySet().stream() + .anyMatch(e -> e.getKey().getName().contains("hazelcast-seda://long") + && Arrays.stream(e.getValue()).anyMatch(f -> "poll".equals(f.getMethodName())))); + long start = System.nanoTime(); + context.getRouteController().stopRoute("long"); + long stopMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + assertTrue(stopMillis < 5000, "Stopping the route took " + stopMillis + " ms"); + + member.getQueue("long").offer("b"); + await().during(3, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(1, member.getQueue("long").size(), + "The item offered after the route stopped was taken from the queue")); + assertEquals(1, mock.getReceivedCounter()); + } + @Test void threadInTheDelayAfterAFailedPollEndsWhenTheRouteStops() throws Exception { // the consumer thread waits in the delay after a failed poll, which runs on the error-delay scheduler @@ -145,6 +169,9 @@ public void configure() { from("hazelcast-seda:idle?hazelcastInstance=#hazelcastInstance").routeId("idle") .to("mock:idle"); + from("hazelcast-seda:long?hazelcastInstance=#hazelcastInstance&pollTimeout=60000").routeId("long") + .to("mock:long"); + from("hazelcast-seda:failing?hazelcastInstance=#failingInstance&onErrorDelay=3000").routeId("failing") .to("mock:failing"); }