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 abb77d3e4aeb CAMEL-26641: camel-core - Aggregate EIP: a suspend should
not force the completion of the groups (#29371)
abb77d3e4aeb is described below
commit abb77d3e4aebc21b04064e555e53981ed9c9327d
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Nov 7 13:31:32 2026 +0530
CAMEL-26641: camel-core - Aggregate EIP: a suspend should not force the
completion of the groups (#29371)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/aggregate-eip.adoc | 4 +-
.../docs/modules/eips/pages/aggregate-eip.adoc | 4 +-
.../processor/aggregate/AggregateProcessor.java | 10 ++
.../aggregator/AggregateSuspendResumeTest.java | 142 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 6 +
5 files changed, 164 insertions(+), 2 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aggregate-eip.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aggregate-eip.adoc
index 9a968b7654f7..ea076943b21b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aggregate-eip.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aggregate-eip.adoc
@@ -267,7 +267,9 @@ for more details.
xref:manual::batch-consumer.adoc[Batch Consumer], which allows you to complete
when all the messages from the batch have been aggregated.
* _forceCompletionOnStop_: Indicates to complete all current
-aggregated exchanges when the context is stopped
+aggregated exchanges when the context (or the route) is stopped. A suspend
+does not force their completion: the groups go on aggregating after the resume
+(completionTimeout and completionInterval still apply).
* _AggregateController_: which allows to use an external source
(`AggregateController` implementation) to complete groups or all groups.
This can be done using Java or JMX API.
diff --git
a/core/camel-core-engine/src/main/docs/modules/eips/pages/aggregate-eip.adoc
b/core/camel-core-engine/src/main/docs/modules/eips/pages/aggregate-eip.adoc
index 9a968b7654f7..ea076943b21b 100644
--- a/core/camel-core-engine/src/main/docs/modules/eips/pages/aggregate-eip.adoc
+++ b/core/camel-core-engine/src/main/docs/modules/eips/pages/aggregate-eip.adoc
@@ -267,7 +267,9 @@ for more details.
xref:manual::batch-consumer.adoc[Batch Consumer], which allows you to complete
when all the messages from the batch have been aggregated.
* _forceCompletionOnStop_: Indicates to complete all current
-aggregated exchanges when the context is stopped
+aggregated exchanges when the context (or the route) is stopped. A suspend
+does not force their completion: the groups go on aggregating after the resume
+(completionTimeout and completionInterval still apply).
* _AggregateController_: which allows to use an external source
(`AggregateController` implementation) to complete groups or all groups.
This can be done using Java or JMX API.
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
index 087991e9681f..e2319d134352 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
@@ -2086,6 +2086,10 @@ public class AggregateProcessor extends
BaseProcessorSupport
@Override
public void prepareShutdown(boolean suspendOnly, boolean forced) {
+ if (suspendOnly) {
+ // only complete the groups when stopping (not when suspending)
+ return;
+ }
// we are shutting down, so force completion if this option was enabled
// but only do this when forced=false, as that is when we have chance
to
// send out new messages to be routed by Camel. When forced=true, then
@@ -2101,6 +2105,12 @@ public class AggregateProcessor extends
BaseProcessorSupport
return true;
}
+ @Override
+ public int getPendingExchangesSize(boolean suspendOnly) {
+ // a suspend keeps the groups (they are not completed), so there is
nothing to wait for
+ return suspendOnly ? 0 : getPendingExchangesSize();
+ }
+
@Override
public int getPendingExchangesSize() {
if (completeAllOnStop) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateSuspendResumeTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateSuspendResumeTest.java
new file mode 100644
index 000000000000..17ecd3621be0
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateSuspendResumeTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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 org.apache.camel.ContextTestSupport;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.BodyInAggregatingStrategy;
+import org.apache.camel.util.StopWatch;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * forceCompletionOnStop and completeAllOnStop complete the groups when the
route or the CamelContext is stopped. A
+ * suspend is not a stop: the groups are kept and go on aggregating after the
resume, and a later stop completes them.
+ */
+class AggregateSuspendResumeTest extends ContextTestSupport {
+
+ @Test
+ void testForceCompletionOnStopKeepsGroupsOnRouteSuspend() throws Exception
{
+ sendTwo("direct:force");
+
+ context.getRouteController().suspendRoute("force");
+ assertEquals(ServiceStatus.Suspended,
context.getRouteController().getRouteStatus("force"));
+ assertEquals(0, getMockEndpoint("mock:force").getReceivedCounter(), "a
suspend completed the group");
+
+ context.getRouteController().resumeRoute("force");
+ expectGroupCompletedAfterResume("direct:force", "mock:force");
+ }
+
+ @Test
+ void testCompleteAllOnStopKeepsGroupsOnRouteSuspend() throws Exception {
+ sendTwo("direct:all");
+
+ // the shutdown timeout of the tests is 10 seconds
+ StopWatch watch = new StopWatch();
+ context.getRouteController().suspendRoute("all");
+ assertTrue(watch.taken() < 5000, "the suspend waited for the groups to
complete: " + watch.taken() + " ms");
+ assertEquals(ServiceStatus.Suspended,
context.getRouteController().getRouteStatus("all"));
+ assertEquals(0, getMockEndpoint("mock:all").getReceivedCounter(), "a
suspend completed the group");
+
+ context.getRouteController().resumeRoute("all");
+ expectGroupCompletedAfterResume("direct:all", "mock:all");
+ }
+
+ @Test
+ void testForceCompletionOnStopKeepsGroupsOnCamelContextSuspend() throws
Exception {
+ // control: a CamelContext suspend did not complete the groups before
either (the context is not started)
+ sendTwo("direct:force");
+
+ context.suspend();
+ assertEquals(0, getMockEndpoint("mock:force").getReceivedCounter(), "a
suspend completed the group");
+
+ context.resume();
+ expectGroupCompletedAfterResume("direct:force", "mock:force");
+ }
+
+ @Test
+ void testCompleteAllOnStopKeepsGroupsOnCamelContextSuspend() throws
Exception {
+ sendTwo("direct:all");
+
+ // the shutdown timeout of the tests is 10 seconds
+ StopWatch watch = new StopWatch();
+ context.suspend();
+ assertTrue(watch.taken() < 5000, "the suspend waited for the groups to
complete: " + watch.taken() + " ms");
+ assertEquals(0, getMockEndpoint("mock:all").getReceivedCounter(), "a
suspend completed the group");
+
+ context.resume();
+ expectGroupCompletedAfterResume("direct:all", "mock:all");
+ }
+
+ @Test
+ void testForceCompletionOnStopCompletesKeptGroupsOnStopAfterSuspend()
throws Exception {
+ expectGroupsCompletedOnStopAfterSuspend("direct:force", "force",
"mock:force");
+ }
+
+ @Test
+ void testCompleteAllOnStopCompletesKeptGroupsOnStopAfterSuspend() throws
Exception {
+ expectGroupsCompletedOnStopAfterSuspend("direct:all", "all",
"mock:all");
+ }
+
+ private void expectGroupsCompletedOnStopAfterSuspend(String uri, String
routeId, String mockUri) throws Exception {
+ MockEndpoint mock = getMockEndpoint(mockUri);
+ mock.expectedBodiesReceived("A+B");
+
+ sendTwo(uri);
+ context.getRouteController().suspendRoute(routeId);
+ assertEquals(0, mock.getReceivedCounter(), "a suspend completed the
group");
+ context.getRouteController().stopRoute(routeId);
+
+ mock.assertIsSatisfied();
+ }
+
+ private void sendTwo(String uri) {
+ template.sendBodyAndHeader(uri, "A", "id", "foo");
+ template.sendBodyAndHeader(uri, "B", "id", "foo");
+ }
+
+ private void expectGroupCompletedAfterResume(String uri, String mockUri)
throws Exception {
+ MockEndpoint mock = getMockEndpoint(mockUri);
+ mock.expectedBodiesReceived("A+B+C");
+
+ template.sendBodyAndHeader(uri, "C", "id", "foo");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:force").routeId("force")
+ .aggregate(header("id"), new
BodyInAggregatingStrategy())
+ .completionSize(3).forceCompletionOnStop()
+ .to("mock:force");
+
+ from("direct:all").routeId("all")
+ .aggregate(header("id"), new
BodyInAggregatingStrategy())
+ .completionSize(3).completeAllOnStop()
+ .to("mock:all");
+ }
+ };
+ }
+}
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 fa61d8a51871..d66a5266da36 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
@@ -333,6 +333,12 @@ let such a failure skip the dead letter channel or
`onException` as before; a pl
same handling. Better, fix the processor to set the exception on the exchange
and call the callback, as an
`AsyncProcessor` must.
+=== camel-core - Aggregate EIP no longer completes the groups when the route
is suspended
+
+With `forceCompletionOnStop` or `completeAllOnStop`, suspending the route no
longer completes the aggregated groups.
+They are kept and completed when the route or Camel is stopped. To send them
out before a suspend, use
+`forceCompletionOfAllGroups` (`AggregateController` or JMX), or stop the route
instead.
+
=== camel-core - ThrottlingExceptionRoutePolicy counts the failures within the
failure window
The circuit of `ThrottlingExceptionRoutePolicy` now opens when
`failureThreshold` failures occurred within the last
