gnodet-bot commented on code in PR #28147: URL: https://github.com/apache/camel/pull/28147#discussion_r4370208795
########## core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffCallerRouteTest.java: ########## @@ -0,0 +1,143 @@ +/* + * 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.errorhandler; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.CamelExecutionException; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.RouteStoppingException; +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.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * CAMEL-25823: when route B is force-stopped while an inflight exchange is inside a nested EIP (split inside + * multicast), the exchange is marked as interrupted after the cut-off WARN is logged. A calling route A that is still + * running and passed the same exchange into route B over {@code direct} then also skips its {@code onException}, + * redelivery, and dead letter channel — the same behaviour as when {@code DirectProducer} is cut off while waiting for + * a consumer (CAMEL-25502). This test pins down that behaviour: the DLC mock is NOT called when the nested-EIP cut-off + * propagates back to the caller. + * + * <p> + * Design: a {@code .process()} inside route B's nested EIP counts down an {@code entered} latch and then blocks + * indefinitely ({@code Thread.sleep(MAX_VALUE)}). The test waits for {@code entered}, force-stops route B (setting its + * error-handler wrappers to stopped state), and then stops route A's seda consumer, which shuts down its thread-pool + * and interrupts the blocked thread. The processor converts the {@code InterruptedException} to a + * {@code RouteStoppingException}. Route B's nested stopped error-handler wrappers then invoke {@code runNotAllowed()}: + * the inner wrapper sees the exception already set and logs the cut-off WARN, then marks the shared exchange as + * interrupted. Route A's error handler then sees the interrupted flag and skips redelivery, {@code onException}, and + * the dead letter channel. + */ +public class RouteStopCutOffCallerRouteTest extends ContextTestSupport { + + private final CountDownLatch entered = new CountDownLatch(1); + + /** + * Route A has a dead letter channel and sends to route B over {@code direct}. Route B is stopped inside a split + * nested in a multicast while a {@code .process()} is blocked. Because the cut-off happens inside a nested EIP of + * route B, {@code setInterrupted(true)} is set on the shared exchange. Route A's error handler then sees the + * interrupted flag and skips redelivery, {@code onException}, and the dead letter channel. + */ + @Test + public void callerRouteDlcNotCalledOnNestedEipCutOff() throws Exception { + MockEndpoint dlc = getMockEndpoint("mock:dlc"); + dlc.expectedMessageCount(0); + + context.getShutdownStrategy().setTimeout(1); + context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS); + + // Route A sends synchronously to route B via direct (same exchange object). + // The Future lets us wait for the exchange to fully propagate back. + Future<Object> future = template.asyncRequestBody("seda:caller", "x", Object.class); + + // Wait until the processor inside route B's nested EIP has entered and is blocking. + entered.await(5, TimeUnit.SECONDS); Review Comment: 🔧 **Nit: unchecked `await` return value** — if the latch times out (returns `false`), the test continues and stops routes before the processor has entered, producing a cryptic assertion failure rather than a clear timeout message. ```suggestion assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not enter within 5 seconds"); ``` (Also add `import static org.junit.jupiter.api.Assertions.assertTrue;`) ########## core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffCallerRouteTest.java: ########## @@ -0,0 +1,143 @@ +/* + * 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.errorhandler; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.CamelExecutionException; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.RouteStoppingException; +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.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * CAMEL-25823: when route B is force-stopped while an inflight exchange is inside a nested EIP (split inside + * multicast), the exchange is marked as interrupted after the cut-off WARN is logged. A calling route A that is still + * running and passed the same exchange into route B over {@code direct} then also skips its {@code onException}, + * redelivery, and dead letter channel — the same behaviour as when {@code DirectProducer} is cut off while waiting for + * a consumer (CAMEL-25502). This test pins down that behaviour: the DLC mock is NOT called when the nested-EIP cut-off + * propagates back to the caller. + * + * <p> + * Design: a {@code .process()} inside route B's nested EIP counts down an {@code entered} latch and then blocks + * indefinitely ({@code Thread.sleep(MAX_VALUE)}). The test waits for {@code entered}, force-stops route B (setting its + * error-handler wrappers to stopped state), and then stops route A's seda consumer, which shuts down its thread-pool + * and interrupts the blocked thread. The processor converts the {@code InterruptedException} to a + * {@code RouteStoppingException}. Route B's nested stopped error-handler wrappers then invoke {@code runNotAllowed()}: + * the inner wrapper sees the exception already set and logs the cut-off WARN, then marks the shared exchange as + * interrupted. Route A's error handler then sees the interrupted flag and skips redelivery, {@code onException}, and + * the dead letter channel. + */ +public class RouteStopCutOffCallerRouteTest extends ContextTestSupport { + + private final CountDownLatch entered = new CountDownLatch(1); + + /** + * Route A has a dead letter channel and sends to route B over {@code direct}. Route B is stopped inside a split + * nested in a multicast while a {@code .process()} is blocked. Because the cut-off happens inside a nested EIP of + * route B, {@code setInterrupted(true)} is set on the shared exchange. Route A's error handler then sees the + * interrupted flag and skips redelivery, {@code onException}, and the dead letter channel. + */ + @Test + public void callerRouteDlcNotCalledOnNestedEipCutOff() throws Exception { + MockEndpoint dlc = getMockEndpoint("mock:dlc"); + dlc.expectedMessageCount(0); + + context.getShutdownStrategy().setTimeout(1); + context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS); + + // Route A sends synchronously to route B via direct (same exchange object). + // The Future lets us wait for the exchange to fully propagate back. + Future<Object> future = template.asyncRequestBody("seda:caller", "x", Object.class); + + // Wait until the processor inside route B's nested EIP has entered and is blocking. + entered.await(5, TimeUnit.SECONDS); + + // 1. Stop route B: sets MulticastProcessor and its nested error-handler wrappers to stopped + // state (isStoppingOrStopped() returns true). The thread is still blocked in the processor. + context.getRouteController().stopRoute("nested-b"); + + // 2. Stop route A: after the shutdown timeout (1 s) its seda executor calls shutdownNow(), + // which interrupts the blocked thread. The processor converts InterruptedException to + // RouteStoppingException; route B's stopped wrappers then invoke runNotAllowed(): + // the outer wrapper logs the cut-off WARN and marks the exchange as interrupted. + // Route A's DLC error handler skips redelivery and the DLC because interrupted=true. + context.getRouteController().stopRoute("caller-a"); + + // Wait for the exchange to finish. It should complete with a RouteStoppingException + // (route A's runInterrupted() keeps the exception and fires the callback without the DLC). + ExecutionException ex = null; + try { + future.get(10, TimeUnit.SECONDS); + } catch (ExecutionException e) { + ex = e; + } Review Comment: 🔧 **Nit: `TimeoutException` not caught** — `future.get(10, SECONDS)` can throw `TimeoutException` (checked, propagates as-is through `throws Exception`). If the exchange hangs — the most likely regression mode here — the test fails with a `TimeoutException` stack trace rather than a clear diagnostic. Catch it explicitly: ```suggestion ExecutionException ex = null; try { future.get(10, TimeUnit.SECONDS); } catch (ExecutionException e) { ex = e; } catch (java.util.concurrent.TimeoutException e) { fail("Exchange did not complete within 10 seconds — likely a hang in the error handler chain", e); } ``` -- 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]
