gnodet-bot commented on code in PR #28147:
URL: https://github.com/apache/camel/pull/28147#discussion_r4457914450
##########
core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogTest.java:
##########
@@ -64,18 +72,38 @@ public void theCutOffIsOneWarnLine() throws Exception {
+ (event.getThrown() != null ? " THROWN" : ""));
}
});
- context.getShutdownStrategy().setTimeout(1);
+ // Use a bounded shutdown timeout so that stopRoute force-stops the
route (stopping its
+ // error handlers) after the timeout, since the inflight exchange will
not complete on its own.
+ // Using direct: (not seda:) ensures the route stop does not interrupt
the processing thread.
+ context.getShutdownStrategy().setTimeout(5);
context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
- context.getInflightRepository().setInflightBrowseEnabled(true);
+ // Send the exchange asynchronously; capture it via a future so we can
wait for it to complete
+ CompletableFuture<Exchange> future =
template.asyncSend("direct:start", exchange -> {
+ exchange.getIn().setBody(List.of("A"));
+ });
- template.sendBody("seda:start", List.of("A", "B"));
- // the exchange waits for a consumer on direct:shipment, as after a
save that sends to a route not written yet
- await().atMost(5, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().browse().stream()
- .anyMatch(e -> "shipment".equals(e.getNodeId())));
+ // Wait for the processor inside the split to enter, then stop the
route while the
+ // exchange is inflight inside the split. The stopped split error
handler will call
+ // runNotAllowed(), set RouteStoppingException, and the outer stopped
handler will log
+ // the cut-off WARN.
+ assertTrue(entered.await(5, TimeUnit.SECONDS), "Processor did not
enter within 5 seconds");
context.getRouteController().stopRoute("slow");
+ // Release the processor after the route is stopped; it exits
normally, and the
+ // stopped error handler chain emits the WARN and completes the
exchange.
+ release.countDown();
+
+ // Wait for the exchange to complete before asserting the log messages
+ Exchange result = null;
+ try {
+ result = future.get(10, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ org.junit.jupiter.api.Assertions.fail(
+ "Exchange did not complete within 10 seconds — likely a
hang in the error handler chain", e);
+ }
Review Comment:
**Nit: use the static import instead of fully-qualified
`org.junit.jupiter.api.Assertions.fail`.**
The other test files in this PR already import `fail` statically. Add to
this file:
```suggestion
import static org.junit.jupiter.api.Assertions.fail;
```
And at line 101:
```java
fail("Exchange did not complete within 10 seconds — likely a
hang in the error handler chain", e);
```
##########
core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogNestedTest.java:
##########
@@ -0,0 +1,253 @@
+/*
+ * 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.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.log.ConsumingAppender;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.junit.jupiter.api.AfterEach;
+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;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * CAMEL-25823: verifies that a route-stop cut-off produces exactly one WARN
line even when the exchange passes through
+ * multiple nested stopped error handlers. Two scenarios are tested:
+ *
+ * <ol>
+ * <li>{@code filter} — a filter wrapping a {@code .process()} step gives the
exchange two stopped error-handler
+ * channels in route B (the {@code .process()} step's channel and the filter's
channel) plus route A's running channel.
+ * The {@code setInterrupted(true)} marker set by the filter's stopped channel
prevents route A's running handler from
+ * logging the same cut-off again.</li>
+ * <li>{@code multicast+split} — a {@code split} nested inside a {@code
multicast} creates a sub-exchange copy.
+ * {@code MulticastProcessor.doDone()} propagates the {@code interrupted} flag
from the sub-exchange to the original so
Review Comment:
**Nit (carried over): comment credits `doDone()` with the propagation, but
the reactive path goes through `taskCallback`.**
`doDone()` covers only the `stopOnException` early-return path. For the
normal reactive path (no `stopOnException`), the propagation happens in
`MulticastReactiveTask.taskCallback`. Suggested correction for all four
occurrences in this file (lines 53, 142, 169, 233) and line 147 in
`RouteStopCutOffCallerRouteTest.java`:
> `taskCallback` (reactive path) propagates `CUTOFF_WARN_LOGGED` and
`interrupted` from the sub-exchange to the original immediately before
`exchangeResult.accept()`. `doDone()` covers the `stopOnException` early-return
path.
##########
core/camel-core/src/test/java/org/apache/camel/processor/errorhandler/RouteStopCutOffLogTest.java:
##########
@@ -126,10 +154,22 @@ protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@Override
public void configure() {
- from("seda:start").routeId("slow")
+ from("direct:start").routeId("slow")
.split(body())
- .log("Picked ${body}")
- .to("direct:shipment").id("shipment")
+ .log("Picked ${body}")
+ // Block here so the test can stop the route while an
exchange is inflight
+ // inside the split. When the route is stopped, the
stopped error-handler
+ // chain emits exactly one cut-off WARN.
+ .process(exchange -> {
+ entered.countDown();
+ try {
+ if (!release.await(30, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("release
latch timed out");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ })
.end()
.to("mock:result");
}
Review Comment:
⚠️ **Blocking: original DirectProducer / `seda + direct:shipment` scenario
no longer covered.**
The PR replaces the entire test body with the latch pattern. Main's version
(`seda:start` → `split().to("direct:shipment")`) specifically tests the
CAMEL-25502 scenario: `DirectProducer` sets `interrupted=true` on the exchange
*before* the error handler's `runNotAllowed()` runs, which is exactly why
`!isInterrupted()` was wrong as the gate — the WARN was silently suppressed on
this path.
The new latch pattern stops the route *before* the split item reaches
`DirectProducer`, so `DirectProducer` never runs and the `interrupted` pre-set
never happens. This means there is no test coverage for the scenario this PR
claims to fix.
Per davsclaus's request: keep main's version unchanged and add the latch
scenario as a new `@Test` method (`theCutOffIsOneWarnLineViaLatch()` or
similar). This way both paths are covered:
- Main's `seda + direct:shipment` → validates WARN is emitted when
`DirectProducer` pre-sets `interrupted=true`
- New latch method → validates WARN is emitted when the split processor is
stopped mid-execution
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/MulticastProcessor.java:
##########
@@ -705,6 +706,18 @@ public void run() {
return;
}
+ // Propagate the interrupted flag and the
cut-off-WARN-logged marker from the
+ // sub-exchange to the original exchange immediately,
before the outer error handler's
+ // callback fires. Only propagate when a stopped error
handler in a nested EIP
+ // (e.g. split inside multicast) already logged the
cut-off WARN -- identified by
+ // the CUTOFF_WARN_LOGGED property set by
RedeliveryErrorHandler.runNotAllowed().
+ // If interrupted was set by DirectProducer without
logging a WARN, we must NOT
+ // propagate: the outer error handler must still log
its WARN for this cut-off.
+ if
(exchange.getProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, false,
Boolean.class)) {
+
original.getExchangeExtension().setInterrupted(true);
+
original.setProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED, Boolean.TRUE);
Review Comment:
⚠️ **Blocking: `MulticastTransactedTask.doRun()` is still missing the
`CUTOFF_WARN_LOGGED` propagation.**
`MulticastReactiveTask.taskCallback` (this hunk, line 716-718) and
`doDone()` (line 1019-1021) both received the marker-guarded copy. But
`MulticastTransactedTask.doRun()` submits via the same
`completion.submit(exchangeResult -> { exchangeResult.accept(exchange); ... })`
at line 855-857 — without any propagation block. In a transacted route with a
split inside a multicast, the outer stopped error handler never sees
`CUTOFF_WARN_LOGGED=true` on the sub-exchange and logs a second WARN.
The upgrade guide says "in any nested EIP" — `MulticastTransactedTask` must
match. Add the same block to its completion lambda:
```java
completion.submit(exchangeResult -> {
// Propagate CUTOFF_WARN_LOGGED from sub-exchange to original before
outer error
// handler's callback fires, same as MulticastReactiveTask.taskCallback.
if (exchange.getProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED,
false, Boolean.class)) {
original.getExchangeExtension().setInterrupted(true);
original.setProperty(RedeliveryErrorHandler.CUTOFF_WARN_LOGGED,
Boolean.TRUE);
}
// accept the exchange as a result
exchangeResult.accept(exchange);
// aggregate exchanges if any
aggregate();
});
```
--
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]
