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 466981910915 CAMEL-26603: camel-core - Aggregate EIP:
forceCompletionOnStop loses the messages that a deferred consumer routes during
the graceful shutdown (#29367)
466981910915 is described below
commit 4669819109152914bbd157b3f1eed153f39ef5ad
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Nov 7 13:31:14 2026 +0530
CAMEL-26603: camel-core - Aggregate EIP: forceCompletionOnStop loses the
messages that a deferred consumer routes during the graceful shutdown (#29367)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/engine/DefaultShutdownStrategy.java | 22 ++++
...egateForceCompletionOnStopLateMessagesTest.java | 119 +++++++++++++++++++++
2 files changed, 141 insertions(+)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
index ae8df13f0546..fbf510891bf5 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
@@ -776,6 +776,28 @@ public class DefaultShutdownStrategy extends
ServiceSupport implements ShutdownS
}
}
+ if (!suspendOnly && !deferredConsumers.isEmpty()) {
+ if (timeoutOccurred.get()) {
+ // the shutdown timed out while the deferred consumers
were prepared (such as a seda consumer
+ // draining its queue), and the thread that called it
shuts down the routes now
+ LOG.debug("Timeout occurred during graceful shutdown, the
shutdown task stops now");
+ return;
+ }
+ // the deferred consumers have routed messages while this task
waited, so prepare the services again
+ // (such as an aggregator with forceCompletionOnStop
completing the groups of these messages) while
+ // the routes can still process them, as stopping a consumer
stops the processors of its route
+ boolean forced =
context.getShutdownStrategy().isForceShutdown();
+ boolean suppress =
context.getShutdownStrategy().isSuppressLoggingOnTimeout();
+ for (RouteStartupOrder order : routes) {
+ for (Service service : order.getServices()) {
+ if (service instanceof Consumer) {
+ continue;
+ }
+ prepareShutdown(service, false, forced, true,
suppress);
+ }
+ }
+ }
+
// now all messages has been completed then stop the deferred
consumers
for (ShutdownDeferredConsumer deferred : deferredConsumers) {
if (timeoutOccurred.get()) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateForceCompletionOnStopLateMessagesTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateForceCompletionOnStopLateMessagesTest.java
new file mode 100644
index 000000000000..eb50c40295bb
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateForceCompletionOnStopLateMessagesTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.processor.aggregator;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+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.builder.AggregationStrategies;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+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;
+
+/**
+ * forceCompletionOnStop completes the groups when the route stops, before the
graceful shutdown waits for the inflight
+ * and pending exchanges. The consumer (seda here) keeps consuming during that
wait, and the messages it routes then
+ * must be completed too, and not be lost.
+ */
+public class AggregateForceCompletionOnStopLateMessagesTest extends
ContextTestSupport {
+
+ private final CountDownLatch blocked = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Test
+ public void testMessagesConsumedWhileStoppingAreCompleted() throws
Exception {
+ doTestMessagesConsumedWhileStopping(false);
+ }
+
+ @Test
+ public void testMessagesConsumedWhileCamelContextStopsAreCompleted()
throws Exception {
+ doTestMessagesConsumedWhileStopping(true);
+ }
+
+ private void doTestMessagesConsumedWhileStopping(boolean stopContext)
throws Exception {
+ MockEndpoint out = getMockEndpoint("mock:out");
+
+ NotifyBuilder first = new NotifyBuilder(context).whenDone(1).create();
+ template.sendBody("seda:in", "m0");
+ assertTrue(first.matches(10, TimeUnit.SECONDS));
+
+ // m1 is in flight (blocked before the aggregator), and m2 and m3 wait
in the seda queue
+ template.sendBody("seda:in", "m1");
+ assertTrue(blocked.await(10, TimeUnit.SECONDS));
+ template.sendBody("seda:in", "m2");
+ template.sendBody("seda:in", "m3");
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> stop = executor.submit(() -> {
+ if (stopContext) {
+ context.stop();
+ } else {
+ context.getRouteController().stopRoute("agg");
+ }
+ return null;
+ });
+ // the stop completes the group of m0 first, while m1 is still in
flight
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
out.getReceivedCounter() >= 1);
+ release.countDown();
+ stop.get(20, TimeUnit.SECONDS);
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ }
+
+ List<String> bodies = new ArrayList<>();
+ for (Exchange e : out.getReceivedExchanges()) {
+ for (Object o : e.getMessage().getBody(List.class)) {
+ bodies.add((String) o);
+ }
+ }
+ bodies.sort(null);
+ assertEquals(List.of("m0", "m1", "m2", "m3"), bodies);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:in").routeId("agg")
+ .process(e -> {
+ if
("m1".equals(e.getMessage().getBody(String.class))) {
+ blocked.countDown();
+ assertTrue(release.await(10,
TimeUnit.SECONDS));
+ }
+ })
+ .aggregate(constant(true),
AggregationStrategies.groupedBody()).completionSize(100)
+ .forceCompletionOnStop()
+ .to("mock:out");
+ }
+ };
+ }
+}
