gnodet-bot commented on code in PR #26797:
URL: https://github.com/apache/camel/pull/26797#discussion_r4089026908
##########
components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaConsumer.java:
##########
@@ -173,10 +178,11 @@ protected void doRun() {
// do not poll if we are suspended or starting again after resuming
if (isSuspending() || isSuspended() || isStarting()) {
- if (shutdownPending && queue.isEmpty()) {
+ if (shutdownPending) {
Review Comment:
⚠️ **Semantic overreach — `isStarting()` should NOT break on
`shutdownPending` without draining.**
The condition `isSuspending() || isSuspended() || isStarting()` bundles
three states. Removing the `queue.isEmpty()` guard is correct for the suspended
states, but `isStarting()` is a different case: it covers the brief window when
a consumer is transitioning START→STARTED (e.g., re-starting after a
`resumeRoute()`). A consumer in that state CAN poll — it just isn't yet — so
the old behaviour (wait for the queue to drain before breaking) was actually
appropriate for `isStarting()`. After this change, a shutdown racing with a
consumer restart will drop pending messages that the consumer would have
processed.
Suggest splitting the condition:
```suggestion
if (shutdownPending && !isStarting()) {
```
Alternatively, pull the `isStarting()` branch into its own `else if` with
the original `queue.isEmpty()` guard. The key invariant: only break out early
when the consumer genuinely cannot make progress (suspended), not when it
momentarily hasn't started polling yet.
##########
core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSuspendedRouteWithPendingStopTest.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+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.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Stopping a suspended seda route must not wait for the messages sent to it
while it was suspended, as a suspended
+ * consumer does not consume them.
+ */
+class SedaSuspendedRouteWithPendingStopTest extends ContextTestSupport {
+
+ private final CountDownLatch processing = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Test
+ void testStopSuspendedRouteWithPendingMessages() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("X");
+
+ // keep the consumer thread busy with X while the consumer is
suspended (as a route policy such as
+ // ThrottlingInflightRoutePolicy does), so it does not poll the queue
while A, B and C are sent
+ template.sendBody("seda:start", "X");
+ assertTrue(processing.await(10, TimeUnit.SECONDS), "X should be
processed");
+ SedaEndpoint seda = (SedaEndpoint)
context.getRoute("foo").getEndpoint();
+ ((SedaConsumer) context.getRoute("foo").getConsumer()).suspend();
+
+ template.sendBody("seda:start", "A");
+ template.sendBody("seda:start", "B");
+ template.sendBody("seda:start", "C");
+ release.countDown();
+ mock.assertIsSatisfied();
Review Comment:
**Minor — direct `consumer.suspend()` diverges slightly from the
`RoutePolicy` path being documented.**
The PR description says this fix addresses the `RoutePolicy`-suspended case
(e.g., `ThrottlingInflightRoutePolicy`). A `RoutePolicy` calls
`routeController.suspendRoute()` (or `route.suspend()` via the controller),
which fires lifecycle events through `DefaultRouteController`. This test calls
`consumer.suspend()` directly, which bypasses the route lifecycle and doesn't
fire `RoutePolicy.onSuspend()` events. It still exercises the `isSuspended()`
code path in `SedaConsumer`, so it covers the fix — but the comment's
description of it as "as a `RoutePolicy` does" isn't quite right.
Not blocking, but worth aligning the test comment to say what it actually
does (suspend the consumer directly to simulate the post-policy state), or add
a second test variant that goes through `suspendRoute()`.
##########
core/camel-core/src/test/java/org/apache/camel/component/seda/SedaSuspendedRouteWithPendingStopTest.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+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.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Stopping a suspended seda route must not wait for the messages sent to it
while it was suspended, as a suspended
+ * consumer does not consume them.
+ */
+class SedaSuspendedRouteWithPendingStopTest extends ContextTestSupport {
+
+ private final CountDownLatch processing = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Test
+ void testStopSuspendedRouteWithPendingMessages() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("X");
+
+ // keep the consumer thread busy with X while the consumer is
suspended (as a route policy such as
+ // ThrottlingInflightRoutePolicy does), so it does not poll the queue
while A, B and C are sent
+ template.sendBody("seda:start", "X");
+ assertTrue(processing.await(10, TimeUnit.SECONDS), "X should be
processed");
+ SedaEndpoint seda = (SedaEndpoint)
context.getRoute("foo").getEndpoint();
+ ((SedaConsumer) context.getRoute("foo").getConsumer()).suspend();
+
+ template.sendBody("seda:start", "A");
+ template.sendBody("seda:start", "B");
+ template.sendBody("seda:start", "C");
+ release.countDown();
+ mock.assertIsSatisfied();
+
+ // abort the stop if the graceful shutdown times out
+ boolean stopped = context.getRouteController().stopRoute("foo", 10,
TimeUnit.SECONDS, true);
+ assertTrue(stopped, "Route should be stopped without waiting for the
shutdown timeout");
+ assertFalse(context.getShutdownStrategy().isTimeoutOccurred());
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("foo"));
+
+ // the suspended consumer did not process the messages, they are kept
on the queue
+ assertEquals(3, seda.getQueue().size());
+
+ // and they are processed when the route is started again
+ mock.reset();
+ mock.expectedBodiesReceived("A", "B", "C");
+ context.getRouteController().startRoute("foo");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testStopContextWithSuspendedRoute() throws Exception {
+ context.getRouteController().suspendRoute("foo");
+
+ template.sendBody("seda:start", "A");
+ template.sendBody("seda:start", "B");
+
+ context.getShutdownStrategy().setTimeout(10);
+ context.stop();
+ assertFalse(context.getShutdownStrategy().isTimeoutOccurred(),
"Graceful shutdown should not time out");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
Review Comment:
**Missing assertion — the invariant for messages after `context.stop()` is
not verified.**
`testStopContextWithSuspendedRoute` asserts only that the timeout did not
occur. It sends A and B while the consumer is suspended, then stops the
context, but never checks what happened to A and B. If
`purgeWhenStopping=false` (the default), they should remain in the queue; if
`purgeWhenStopping=true`, they should be purged. The test silently passes
either way, which means a regression that drops the messages without a timeout
would also pass.
Suggest adding a queue-size assertion (or at minimum a note on why one is
omitted, since the queue is GC'd when the context stops and may not be
accessible):
```suggestion
assertFalse(context.getShutdownStrategy().isTimeoutOccurred(),
"Graceful shutdown should not time out");
// A and B were queued while the consumer was suspended; they are
kept on the queue (not purged)
// because purgeWhenStopping defaults to false. The queue is owned
by the endpoint and lives
// until the context is fully stopped, so we cannot assert its size
here — the endpoint is torn down
// before this line. The absence of a timeout is the observable
invariant.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
