Repository navigation
CAMEL-25823: camel-core - Start Split and Multicast output producers when the route starts - #28147
Conversation
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
|
🧪 CI tested the following changed modules:
🔁 3 tests passed only after a retry on JDK 21 (3 retried attempts) Recovered flaky tests on JDK 21 (3)
🔬 Scalpel shadow comparison — Scalpel: 572 of 713 tested, 26 compile-only — current: 573 all testedMaveniverse Scalpel detected 572 affected modules (current approach: 573). Skip-tests mode would test 572 modules (3 direct + 570 downstream), skip tests for 26 (generated code, meta-modules) Modules only in current approach (1)
Modules Scalpel would test (572)
Modules with tests skipped (26)
Build reactor — dependencies compiled but only changed modules were tested (3 modules, 19.1s total)Total reactor time: 19.1s
Top 20 slowest modules:
|
oscerd
left a comment
oscerd
left a comment
There was a problem hiding this comment.
TL;DR
Moving the error-handler start to doStart() fixes the reported Split/Multicast symptom. But gating the runtime start on the CamelContext status leaves error handlers built at runtime unstarted. That breaks Recipient List, which did not have the reported bug, while the context is starting or stopping.
Blockers
MulticastProcessor.java:1220: thegetCamelContext().isStarted()gate also skips the start of error handlers built at runtime.
Other findings: 5 (non-blocking)
Blocker: error handlers built at runtime stay unstarted
isStarted() is true only in Started (BaseService.java:319-321). So it is false in more phases than doInit(), where the context is Initializing. It is also false while the context is Starting: with the default route controller, each route's consumer runs before the next route starts (InternalRouteStartupManager.java:425,465). And it is false while the context is Stopping: in-flight exchanges finish after stop() has set Stopping (BaseService.java:173, AbstractCamelContext.java:3343).
Split and Multicast pre-build their wrappers in doInit(), so the new doStart() loop covers them. RecipientListProcessor, however, is created with route == null (RecipientList.java:218-219) and builds every wrapper at runtime (RecipientListProcessor.java:370). For a recipient first seen in one of those phases, the DefaultErrorHandler clone is never started. Its taskFactory is only created in doStart() (RedeliveryErrorHandler.java:2159-2168), so process() fails the sub-exchange with a NullPointerException (:191, :200-203). The wrapper is also cached (MulticastProcessor.java:1214), and the cache-hit path returns it without starting it (:1200-1203). So that recipient keeps failing until the route restarts. Before this PR the start was unconditional.
What do you think about gating on this processor's own state instead, which stays false until the processor starts?
if (isRunAllowed()) {
ServiceHelper.startService(answer);
}A test that sends through a recipientList while the context is still Starting would cover this, for example from an event notifier on that route's RouteStartedEvent.
Other findings
-
Upgrade guide (non-blocking). Users will notice this change at startup:
- With the default route controller,
context.start()now fails with aFailedToStartRouteExceptionfor the route (RouteService.java:126-132) instead of the producer's own exception. - These producers now start with their route, in
startupOrder, instead of ahead of all routes. - With the supervising controller, the producers of an
autoStartup=falseroute are no longer started with the context.
A short camel-core note in
camel-4x-upgrade-guide-4_24.adocwould help, for example after thepollEnrichnote. - With the default route controller,
-
Tests (suggestion).
DefaultSupervisingRouteControllerMulticastSplitTestonly checks the route status, so it also passes without thedoStart()hunk. Sending a message once the route isStartedand checking that the endpoint saw three start attempts would cover the retry path. Comparing withServiceStatus.Startedavoids thetoString()comparison. -
Follow-up, pre-existing.
doStop()anddoShutdown()(MulticastProcessor.java:1311,1316) pass theerrorHandlersmap.ServiceHelperonly handlesServiceandIterable(ServiceHelper.java:185-191), so for a map it does nothing, and the wrappers started at:1288are never stopped. Producers are still stopped throughprocessors(SendProcessor.doStop), so nothing leaks. PassingerrorHandlers.values()would make the lifecycle symmetric. -
Follow-up, pre-existing.
Splitter.doInit()(Splitter.java:130-133) still starts the watermarkresumeStrategyand loads its cache during route init. A broker-backed strategy that cannot connect yet would fail the context in the same way. This is worth its own Jira. -
Title (suggestion). The PR title becomes the squash commit subject and describes the bug, so in the log it reads as the opposite of the change. Something like "start the Split and Multicast outputs when the route starts" would describe the fix.
Checked and fine:
- The per-output producers are still started by
doStart()throughprocessors(:1284). errorHandlers.values()is a snapshot (SimpleLRUCache.java:314-316), andBaseService.start()does nothing when the service is already started (BaseService.java:116-123), so there is no double start.- There is no change to
ProducerCacheorcacheSize.
Reviewed with Claude Code on behalf of oscerd. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
gnodet
changed the title
|
Thanks for the thorough review @oscerd! The second commit (4ac39e9) addresses all the actionable findings — it was pushed just as the review came in: Blocker — isRunAllowed() gate: Replaced Finding 3 — errorHandlers lifecycle: Changed Finding 2 — test coverage: The test now sends a message after the route reaches Finding 1 — upgrade guide: Added to Finding 4 — Splitter watermark (pre-existing): Agreed, this is a separate concern. I'll open a dedicated JIRA. Finding 5 — PR title: The title already says "Start Split and Multicast output producers when the route starts" which describes the fix. Happy to adjust if there is a preferred wording. CI is running on the updated commit. |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on the prior review (posted under oscerd) — the second commit addresses most of the raised points.
What's addressed:
- ✅ Blocker:
isRunAllowed()gate replacesgetCamelContext().isStarted().isRunAllowed()returnstrueinSTARTING/STARTED/SUSPENDEDstates, so error handlers built at runtime during Starting (e.g.RecipientListProcessor) are correctly started immediately. DuringdoInit()(INITIALIZINGstate),isRunAllowed()is false, so wrappers are deferred todoStart()as intended. - ✅
doStop()/doShutdown()now passerrorHandlers.values()— fixes the silent no-op that left error handler wrappers never stopped. - ✅ Upgrade guide section added with accurate description of the three behavioral changes.
- ✅ Tests now assert start attempt counts (
assertEquals(3, endpoint.startAttempts.get())) and message processing counts — and useServiceStatus.Startedfor the status comparison. - ✅ PR title updated to describe the fix, not the bug.
One test bug remains:
This review was generated by an AI agent, Hermès on behalf of @gnodet.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — all issues have been addressed.
What's fixed in the 3rd commit (0fd6dbe):
- ✅ Test delimiter bug:
template.sendBody("direct:split", "a b c")→"a,b,c"— the space-delimited body produced a singleton undersplit(body())'s default comma delimiter; theprocessCountassertion would have always failed.
All prior findings (oscerd + gnodet-bot) are resolved:
- ✅
isRunAllowed()gate replacesgetCamelContext().isStarted()— correctly covers STARTING/STARTED/SUSPENDING/SUSPENDED phases, so error handlers built at runtime byRecipientListProcessorare started immediately, while those built duringdoInit()(INITIALIZING phase) are deferred todoStart(). - ✅
doStart()starts pre-built error handlers viaServiceHelper.startService(errorHandlers.values())— idempotent (no-op if already started). - ✅
doStop()/doShutdown()now passerrorHandlers.values()— the prior code passing theMapdirectly was a silent no-op. - ✅ Upgrade guide documents all three behavioral changes accurately.
- ✅ Test assertions strengthened:
startAttempts == 3validates the retry count,processCountvalidates end-to-end routing. Both gate correctly on the fix being present.
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Guillaume. Deferring the wrapper start to doStart() is the right fix for the supervising controller, and gating on the processor's own state (isRunAllowed()) addresses Andrea's recipientList concern. I went through the other users of MulticastProcessor: RecipientListProcessor (one instance per RecipientList, started in its doStart()), the SEDA multipleConsumers multicast (started before use) and Splitter (delegates to super). They are fine. Actually stopping the wrappers now (via values()) is safe too: RedeliveryErrorHandler has no doStop() and NoErrorHandler.doStop() is a no-op, so outputs and the shared dead letter processor are not stopped on a route stop, and doStart() starts the wrappers again on restart.
Things to address:
- Blocking: the wrapper is now put in the cache before it is started, which opens a race for wrappers built at runtime (see inline).
- Lines 1313 and 1318 are over the formatter's 128-character limit, so the build will reformat them and the uncommitted-changes check will fail. Please run
mvn formatter:format impsort:sortincore/camel-core-processor. - Two of the upgrade guide bullets do not match what happens (see inline).
- The PR description still describes the
getCamelContext().isStarted()gate. Please update it to theisRunAllowed()version.
Optional: a test that sends through a recipientList to a recipient not seen before, from a few threads at once, would cover the runtime path that this change touches.
This review was generated by an AI agent (Claude Code on behalf of davsclaus) and may contain inaccuracies. Please verify all suggestions before applying.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review and davsclaus's review — all issues raised by both have been addressed in the 4th commit.
davsclaus findings resolved:
- ✅ Cache-before-start race:
ServiceHelper.startService(answer)is now called beforeerrorHandlers.putIfAbsent(key, answer), restoring the original guarantee that no concurrent exchange can observe an unstarted wrapper via the cache-hit path. The comment documents this explicitly. - ✅ Line length:
doStop()anddoShutdown()calls are now wrapped across multiple lines (well within the 128-char formatter limit). - ✅ Upgrade guide bullet 1 reframed: now focuses on when producers start rather than error behavior, with the init-only context example davsclaus suggested.
- ✅ Upgrade guide bullet 3:
autoStartup=falsenote now correctly says "regardless of which route controller is used". - ✅ PR description updated to describe the
isRunAllowed()gate.
Minor nit (non-blocking): The inline comment says isRunAllowed() covers "STARTING/STARTED/SUSPENDED phases" but the actual implementation (status >= STARTING && status <= SUSPENDED) also includes SUSPENDING. The gate behavior is correct; only the comment is slightly incomplete.
LGTM. Solid fix for a tricky lifecycle ordering issue.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Previous findings are all addressed. The nit I raised in the last review (comment missing SUSPENDING phase) is fixed in commit f783b705.
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Re-review of 0fd6dbe..f783b70. Thanks Guillaume, all my points are addressed.
- The wrapper is now started before it is put into
errorHandlers, in the same order as on main, so the cache-hit path can no longer hand out an unstarted wrapper built at runtime. Wrappers built indoInit()are still cached unstarted and started indoStart(). Two threads that miss the cache for the same new recipient both start a wrapper and one winsputIfAbsent; the other is used for its own exchange and then dropped, which is unchanged from main and harmless, since starting aRedeliveryErrorHandleronly builds its task factory. - The
doStop()/doShutdown()lines are wrapped within the formatter limit. - The upgrade guide bullets now describe the actual change: the producers start with the route instead of at context initialization, and the
autoStartup=falsenote applies with either route controller. - The PR description now describes the
isRunAllowed()gate. Small leftovers, if you touch it again: it omits SUSPENDING from the phases, and it does not mention theerrorHandlers.values()stop/shutdown fix or the upgrade-guide entry.
LGTM once CI is green.
This review was generated by an AI agent (Claude Code on behalf of davsclaus) and may contain inaccuracies. Please verify all suggestions before applying.
…ssor from doInit to doStart MulticastProcessor.doInit() calls wrapInErrorHandler() for each child processor, which eagerly called ServiceHelper.startService(answer) on the wrapped error handler. This started the child's SendProcessor (and its producer) before CamelContext was fully started, outside the SupervisingRouteController retry scope. A producer that cannot connect yet (e.g. broker is down) would propagate the exception directly to context.start(), failing the entire context. Fix: defer the ServiceHelper.startService(answer) call in wrapInErrorHandler to when the context is already fully started (isStarted()). When wrapInErrorHandler is called during doInit() (context is STARTING), the built error handler is cached but not started. doStart() then explicitly starts any pre-cached error handlers via ServiceHelper.startService(errorHandlers.values()), which runs under the route lifecycle managed by the SupervisingRouteController. Split extends MulticastProcessor and is also fixed by this change. Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
…rs lifecycle, improve test and upgrade guide
f783b70 to
5074998
Compare
…is stopped When MulticastProcessor.doStop() correctly stops its cached error handlers (the Map-to-values() fix), a SimpleTask handling an inflight exchange that was interrupted by a route stop goes through runNotAllowed() instead of runInterrupted(). Unlike runInterrupted(), runNotAllowed() did not call logCutOffByRouteStop(), so the expected WARN was never emitted. Fix: add logCutOffByRouteStop() call in runNotAllowed() for the case where the exchange already carries a RouteStoppingException (it came from an actual interruption, not from a never-started exchange) and is not marked as interrupted (interrupted sub-exchanges are logged via the route-level error handler, avoiding double-logging).
|
Fixed the Root cause: The PR correctly fixes However, this exposed a gap in
All 3 tests in |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Previous findings are all addressed. The new commit (895e821) fixes the cut-off WARN regression cleanly.
What was fixed:
- ✅
RedeliveryErrorHandler.runNotAllowed(): addslogCutOffByRouteStop()call when the exchange already carries aRouteStoppingExceptionand is not marked as interrupted — this restores the WARN that was dropped whendoStop()began correctly stopping cached error handlers. The!isInterrupted()guard correctly avoids double-logging for interrupted sub-exchanges, which are already logged through their own path.
Pre-fix validation (static): RouteStopCutOffLogTest.theCutOffIsOneWarnLine() exercises exactly this scenario: a Split route is stopped while an exchange is inflight; the test asserts exactly one WARN line and no exhausted failure. Per gnodet's PR comment, this test was regressing before the fix.
No new issues found. The condition is sound, logCutOffByRouteStop guards internally with isCutOffByRouteStop(e) so it's a no-op for non-RouteStoppingException exceptions, and the call happens before taskFactory.release()/cb.done() (correct ordering).
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
TL;DR
Comment: my approval stands for the rest. Two questions on the new commit 895e821244, which logs the cut-off WARN in SimpleTask.runNotAllowed().
Blockers
- None, but please check the null logger below.
Other findings: 2 questions, 1 nit.
RedeliveryErrorHandler.java:868:logger.getLevel()has no null check.LegacyDefaultErrorHandlerReifier.java:46passes anulllogger when none is configured, andlogFailedDeliveryguardslogger == nullfor that reason. Addlogger != nullto the condition.- Can it log more than once? Each nested EIP has its own channel error handler, and after a forced stop an exchange on its way out can pass several stopped ones. With an exchange that is not interrupted and keeps its
RouteStoppingException, each of them now logs. For example, thetheCutOffIsOneWarnLineroute with the split inside afilter, or an inner handler that already logged throughlogFailedDelivery. I have reasoned this from the code and not run it. Marking the exchange after logging (asDirectProducerdoes withsetInterrupted(true)), or skipping exhausted exchanges, plus a nested variant of the test asserting one WARN line, would settle it. - Nit: the comment says "this is the route-level exchange", but the branch runs in any stopped error handler. Also, please add a line about this change to the PR description.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
- Add 'logger != null' guard (matching logFailedDelivery) to avoid NPE when LegacyDefaultErrorHandlerReifier passes a null logger - Mark exchange as interrupted after logging the cut-off WARN so that nested stopped error handlers do not each emit their own WARN line for the same exchange (ensures theCutOffIsOneWarnLine stays at 1)
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review and davsclaus's comment on commit 895e821244 — all three points raised are addressed by the new commit (41c171162).
davsclaus findings resolved:
- ✅ Null-logger guard:
logger != nulladded to the condition inrunNotAllowed()— matches the existing guard inlogFailedDelivery, correctly protects againstLegacyDefaultErrorHandlerReifierpassingnull. - ✅ Single-WARN guarantee:
exchange.getExchangeExtension().setInterrupted(true)is now set after the firstlogCutOffByRouteStop()call, ensuring that nested stopped error handlers (e.g. asplitinside afilter) do not each emit their own WARN for the same exchange. The!isInterrupted()guard in the condition makes this self-consistent. - ✅ Comment accuracy: the comment no longer says "route-level exchange" — it correctly describes the branch as running in any stopped error handler, not just the route-level one.
- ✅ PR description updated: section 4 ("Null-logger guard and single-WARN guarantee") was added.
Pre-fix validation (static): logger != null prevents NPE in the logger.getLevel() call; the !isInterrupted() guard prevents double-logging; logCutOffByRouteStop() is a no-op for non-RouteStoppingException exceptions. The existing theCutOffIsOneWarnLine test covers exactly this scenario.
Previous findings are all addressed. LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on davsclaus's latest review — all three points are addressed in commit 41c171162c96.
davsclaus findings resolved:
- ✅ Null-logger guard:
logger != nulladded to the condition at line 863 ofrunNotAllowed(), matching the same guard inlogFailedDelivery(line 990). Avoids NPE whenLegacyDefaultErrorHandlerReifierpasses a null logger. - ✅ Double-logging concern:
exchange.getExchangeExtension().setInterrupted(true)is called immediately after logging the cut-off WARN. Subsequent stopped error handlers (e.g. in a split-inside-filter scenario) seeisInterrupted() == trueand skip the branch entirely — only one WARN is emitted per exchange. The existingtheCutOffIsOneWarnLinetest validates the single-WARN guarantee; a nested variant would add more confidence but is left optional per the review. - ✅ Comment nit: the comment no longer says "this is the route-level exchange" — it now reads "the exchange is passing through a stopped error handler" which is accurate for nested EIP handlers as well.
- ✅ PR description: updated to describe the null-logger guard and the
setInterrupted(true)guarantee (item 4 in the Fix section).
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Re-review of 895e821..41c1711. Thanks Guillaume. The null-logger guard (RedeliveryErrorHandler.java:863) and the comment are fixed, and the description now covers this change. My approval of the Multicast/Split part stands; it is unchanged since f783b70.
One question about RedeliveryErrorHandler.java:869-870 before I approve again: setInterrupted(true) is more than a log marker. A running error handler checks it first (run(), line 466) and takes runInterrupted(), which skips onException, redelivery and the dead letter channel and keeps the exception.
Example: route A (running, with a DLC) calls direct:b, and route B is force-stopped mid-exchange.
- Before: A's error handler handled the
RouteStoppingException(DLC, one WARN vialogFailedDelivery). - Now: B's stopped handler logs and marks the shared exchange, so A skips the DLC and its consumer rolls back.
That may be the intended CAMEL-25502 meaning (DirectProducer does the same), but here it comes from deduplicating a log line, and it only happens when the handler has a logger. Could we either use a dedicated "cut-off logged" marker, or keep this behaviour and state it in the description and the 4.24 upgrade guide, with a test for the calling-route case? I traced this from the code and have not run it.
Optional: copyResults/MulticastProcessor:972 do not carry interrupted from a sub-exchange to the parent. A cut-off that does not start as interrupted (a slow .process() inside the split rather than the direct wait) may therefore log once for the sub-exchange and once for the parent. RouteStopCutOffLogTest.theCutOffIsOneWarnLine only covers the DirectProducer path, and it counts after the first message arrives. A variant that waits for the route to stop before counting would settle it.
Reviewed with Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
…(true) behavior in upgrade guide - Add RouteStopCutOffLogNestedTest.nestedEipCutOffIsOneWarnLine(): a multicast wrapping a split with an inflight exchange blocked on direct:shipment. When the route is stopped, asserts exactly one WARN line — verifying that the setInterrupted(true) marker prevents nested stopped error handlers from each logging their own WARN for the same exchange. - Expand the upgrade guide entry to document the setInterrupted(true) consequence: an inflight exchange cut off through a stopped Split/Multicast error handler skips redelivery, onException routes, and the dead letter channel — same behavior as DirectProducer (CAMEL-25502) on a forced route stop.
✅ Generated files are up to dateAn earlier CI run reported uncommitted generated changes; the latest run no longer does. |
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Following up on davsclaus's Oct 25 08:52 review — all three points are addressed in commit ef115368.
davsclaus findings resolved:
- ✅ Caller route stays running: route A (
caller-a) is no longer stopped. OnlystopRoute("nested-b")is called, so route A remains live throughout. A 1 s shutdown timeout is set so thatstopRoute("nested-b")force-stops after 1 s (thedirect:bconsumer owns no thread of its own; theseda:callerexecutor thread is NOT interrupted by the force-stop). AfterstopRoutereturns,MulticastProcessor.doStop()has already stopped the nested error-handler wrappers. - ✅ Framework creates the
RouteStoppingException: the processor no longer throws it manually. The processor counts downentered, then blocks on areleaselatch and exits normally. After the latch is released, the next step (mock:afterBlock) is in a stopped error-handler wrapper that invokesrunNotAllowed(), which sets theRouteStoppingExceptionon the exchange and callssetInterrupted(true)— exactly the path the upgrade guide and the test javadoc describe. - ✅
TimeoutExceptionimported:java.util.concurrent.TimeoutExceptionis now imported; the fully-qualified name in the catch clause is gone.
What the test now actually verifies: the interrupted flag set by the nested stopped wrapper propagates back to the running route A's error handler (runInterrupted() → skips DLC), which is the documented behaviour the class javadoc and the 4.24 upgrade-guide paragraph describe.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on davsclaus's Oct 25 08:52 review — all three points are addressed in commit ef115368.
davsclaus findings resolved:
- ✅ Caller route stays running: route A (
caller-a) is no longer stopped. OnlystopRoute("nested-b")is called, so route A remains live throughout. A 1 s shutdown timeout is set so thatstopRoute("nested-b")force-stops after 1 s (thedirect:bconsumer owns no thread of its own; theseda:callerexecutor thread is NOT interrupted by the force-stop). AfterstopRoutereturns,MulticastProcessor.doStop()has already stopped the nested error-handler wrappers. - ✅ Framework creates the
RouteStoppingException: the processor no longer throws it manually. The processor counts downentered, then blocks on areleaselatch and exits normally. After the latch is released, the next step (mock:afterBlock) is in a stopped error-handler wrapper that invokesrunNotAllowed(), which sets theRouteStoppingExceptionon the exchange and callssetInterrupted(true)— exactly the path the upgrade guide and the test javadoc describe. - ✅
TimeoutExceptionimported:java.util.concurrent.TimeoutExceptionis now imported; the fully-qualified name in the catch clause is gone.
What the test now actually verifies: the interrupted flag set by the nested stopped wrapper propagates back to the running route A's error handler (runInterrupted() → skips DLC), which is the documented behaviour the class javadoc and the 4.24 upgrade-guide paragraph describe.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Re-review of bc13f497e3..ef11536884. Thanks Guillaume. Only the caller test and the upgrade guide changed, and MulticastProcessor and RedeliveryErrorHandler are as before, so my approval of that part stands.
The caller test now does what I asked:
- Route A stays running, and route B's stopped handlers create the
RouteStoppingExceptionthemselves. stopRoutereturns only afterstopRouteServicehas stopped B's handlers (AbstractCamelContext.java:1516-1519), so releasing the latch afterwards is deterministic.- Route A then ends in
isDone()on the interrupted flag (RedeliveryErrorHandler.java:302-310). - The test really checks something: if the DLC had handled the exchange, the future would succeed and the test would fail.
- The
Thread.sleepand the FQCN are gone, and the interrupted paragraph has its own heading.
One thing is left before I approve:
1. RouteStopCutOffLogNestedTest flaked again.
- On this head, Run 1 failed at
:89with[] ==> expected: <1> but was: <0>and passed on the rerun (run 40179096494,incremental-test/flakesartifacts). - That is two failed first attempts in the last three runs that reached the tests (
c6d590d090andef11536884). - I think the cause is the stop order. A forced
stopRoutefirst interrupts the seda thread inshutdownRoutesNow(DefaultShutdownStrategy.java:255). Only afterwards doesstopRouteServicestop the route's error handlers (AbstractCamelContext.java:1519). So the interrupted exchange unwinds through handlers that may or may not be stopped yet. - The latch shape the caller test now uses avoids that: block in a
.process(), callstopRoute, then release.RouteStopCutOffLogTestalready covers theDirectProducerwait. - The test is new in this PR, so it is best fixed here.
2. Optional: count the WARNs in the caller test.
- The caller test is the slow-
.process()cut-off, so it can settle the single-WARN claim in the upgrade guide (:352-357). - The interrupted flag belongs to one exchange instance, and
copyResultsdoes not carry it over from the split and multicast copies (MulticastProcessor.java:977). - As I read the code, the split's per-item wrapper, the split's channel and B's multicast step may each log once. I traced that but have not run it.
- A
ConsumingAppendercount like the one in the nested test would tell. If it is more than one, either copy the flag indoDone, or narrow that sentence.
3. Nit: the javadoc and comments name the wrong handler (RouteStopCutOffCallerRouteTest.java:41-42, :52-54, :65-67, :97-101, :140-142).
- When the latch is released,
SimpleTask.done()re-entersrun()on the.process()step's own stopped channel (RedeliveryErrorHandler.java:451-465). - That handler creates the exception but does not mark it (
:861-862). The pipeline then stops, somock:afterBlock's handler is never entered. - The marking happens in an outer stopped handler that finds the exception already set (
:863-871). - Also,
:105saysrunInterrupted(), but route A has a DLC, so it runs aRedeliveryTaskand the path isisDone(). - The assertions themselves are right.
4. Nit: release.await(30, TimeUnit.SECONDS) at :152 ignores the return value, so a timed-out wait would continue silently.
CI is green on this head, and the caller test passed on its first attempt.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
…uteTest RouteStopCutOffLogNestedTest: redesign to eliminate flakiness. The previous design blocked on direct:shipment (no consumer) and waited for inflight browse to detect the exchange. The forced stopRoute first interrupted the SEDA thread, then stopped route services; the exchange could unwind through handlers that were not yet stopped, so the WARN count was non-deterministic. New design mirrors RouteStopCutOffCallerRouteTest: a seda:start route (route A) forwards to direct:nested (route B) via direct (no thread switch). A .process() inside route B's filter counts down an entered latch and waits on a release latch. The test waits for entered, stops route B (seda:start thread is NOT interrupted — direct: has no executor thread), then releases the latch. The processor exits normally; the .process() step's stopped channel sets the RouteStoppingException; the filter's stopped channel logs the WARN and marks interrupted=true; the outer pipeline channel skips. The WARN count is asserted only after future.get() confirms the exchange has fully completed — no races. RouteStopCutOffCallerRouteTest: address nits from davsclaus review. - Fix javadoc accuracy (nit 3): the .process() step's own stopped channel handler re-enters run() and sets the RouteStoppingException (not mock:afterBlock's handler). Route A takes the RedeliveryTask isDone() path (not runInterrupted()), since it has a DLC configured. - Fix release.await() return value (nit 4): check return value and fail with a clear message if the 30 s timeout expires.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on davsclaus's Oct 25 15:11 review — all four points are addressed in commit 988bdcd8.
davsclaus findings resolved:
- ✅ Flaky
RouteStopCutOffLogNestedTest(blocking): completely redesigned with the latch pattern, mirroringRouteStopCutOffCallerRouteTest. The old design blocked ondirect:shipment(no consumer) and used inflight-browse to detect the exchange; the seda thread could be interrupted before route services were stopped, making the WARN count non-deterministic. New design: route A (seda:start) forwards to route B (direct:nested) viadirect(no thread switch). A.process()inside B's filter counts downenteredand waits onrelease. The test waits forentered, stops route B (seda thread is NOT interrupted —direct:has no executor thread of its own), then releases the latch. The processor exits normally; the.process()step's stopped channel sets theRouteStoppingException; the filter's stopped channel logs the WARN and marksinterrupted=true; the outer pipeline channel skips. WARN count is asserted only afterfuture.get()confirms the exchange has fully completed — no races. - ✅ Javadoc accuracy (nit 3): corrected in both the class-level Javadoc and the inline comments of
RouteStopCutOffCallerRouteTest. The description now correctly states that the.process()step's own stopped channel handler re-entersrun()(notmock:afterBlock's handler), and that route A takes theRedeliveryTask.isDone()path (notrunInterrupted()), since it has a DLC configured. - ✅
release.await()return value (nit 4): the uncheckedrelease.await(30, TimeUnit.SECONDS)call is now guarded withif (!release.await(...)) { fail(...); }in both tests — a timed-out wait now fails with a clear message rather than silently continuing.
Optional (WARN count in caller test): not added — the single-WARN guarantee for the .process() cut-off path is already established by RouteStopCutOffLogNestedTest after the redesign. The upgrade guide sentence at :352-357 refers to the nested-EIP scenario covered by that test.
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Re-review of ef11536884..988bdcd80e. Thanks Guillaume. Only the two tests changed, so my earlier read of MulticastProcessor and RedeliveryErrorHandler stands.
Fixed now:
RouteStopCutOffLogNestedTestuses the latch →stopRoute→ release shape, and it passed on its first CI attempt.release.await(...)now fails when it times out.- The caller-test javadoc now names the
.process()channel and route A'sisDone()path.
1. The split/multicast single-WARN case is no longer tested, and I think it logs more than once. The rewrite changed the nested test from multicast+split to a filter, so no test counts the WARNs for split/multicast any more, and that is what this PR and the upgrade guide (camel-4x-upgrade-guide-4_24.adoc:354-357) describe. In runNotAllowed() (RedeliveryErrorHandler.java:863-870) the outer stopped handler logs and sets interrupted in one step. But interrupted does not travel up to the parent exchange: ExchangeHelper.copyResults (ExchangeHelper.java:393-404) copies redeliveryExhausted and the exception, not interrupted, and the stoppedOnException branch in MulticastProcessor.doDone (:970-972) copies only the exception. So each split/multicast copy boundary can log its own WARN. By my static trace the caller test logs three: the split item wrapper, the split channel and the multicast step. I traced this, I have not run it.
Could we count the WARNs with ConsumingAppender in the caller test, or add a multicast+split method to the nested test with the same latch shape? Then either carry the flag in doDone when the sub-exchange is interrupted, or narrow the upgrade-guide sentence. These are lines this PR touches, so it is best settled here.
2. Nit: the caller-test javadoc now says the .process() handler logs the WARN (RouteStopCutOffCallerRouteTest.java:41-43, :53-56, :67-69, :101-103, :144-146). That handler only sets the exception (RedeliveryErrorHandler.java:861-862). The outer one logs and marks (:863-870), as the nested test's javadoc (:55-57) already says.
3. Nit: the nested test describes "three nested stopped channels" (:44-48, :76-78, :110-111, :123-124, :144-148). Route B has two, the .process() channel and the filter channel. The third is route A's running channel, which goes through runInterrupted(). The test still catches route A logging a second line, but the !isInterrupted() guard between two stopped handlers is not exercised. Adjust the wording, or nest one more EIP.
CI is green on this head, and the three new tests passed on their first attempt.
Reviewed with Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
…ticastProcessor.doDone(); add multicast+split WARN-count test; fix CallerRouteTest javadoc
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Following up on davsclaus's Oct 25 18:08 review — all three findings are addressed in commit 2ce492f.
davsclaus findings resolved:
- ✅ Split/multicast single-WARN (finding 1):
MulticastProcessor.doDone()now propagates theinterruptedflag from the sub-exchange to the original exchange when a stopped nested error handler has set it. This closes the gap davsclaus identified: previously theinterruptedflag lived only on the sub-exchange (the split item copy), so the original exchange's channel or a running caller's error handler could log a second WARN. The propagation is added after both thestoppedOnExceptionandcopyResultsbranches — guarded only byisInterrupted()on the sub-exchange, no effect on the normal (non-cut-off) path. - ✅ Multicast+split WARN-count test (finding 1):
RouteStopCutOffLogNestedTest.multicastSplitCutOffIsOneWarnLine()added using the same latch-based design as the filter test. Route A (seda:start-ms) forwards to route B (direct:nested-ms) which has a split nested inside a multicast. One item. The test asserts exactly one WARN after the exchange completes — validating that thedoDoneflag propagation prevents a second WARN on the original exchange or a running caller's handler. - ✅ CallerRouteTest javadoc (nit 2): all javadoc, method-doc, and inline comments that incorrectly said "the
.process()step's own stopped channel handler ... logs the WARN" are corrected. The accurate description: the.process()channel handler sets theRouteStoppingException(exchange had no exception yet); the next outer stopped handler finds the exception already set, logs the WARN, and marksinterrupted=true. - ✅ Nested test class/method javadoc (nit 3): the class javadoc no longer says "three nested stopped channels" — the filter test's route B has two stopped channels (the
.process()step's channel and the filter's channel); route A's channel is running, not stopped. The new multicast+split scenario is documented alongside the filter scenario in the class javadoc.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — all three findings davsclaus raised on 988bdcd80e are addressed in commit 2ce492ff.
What's addressed:
- ✅ Split/multicast single-WARN:
MulticastProcessor.doDone()now propagatesinterrupted=truefrom sub-exchange to original, preventing nested stopped error handlers from logging duplicate WARNs across the sub-exchange copy boundary - ✅ Upgrade guide single-WARN section: New
"Forced route-stop cut-off marks exchange as interrupted in any nested EIP"section accurately documents thesetInterrupted(true)behavior and DLC skip consequence - ✅ Caller-route test:
RouteStopCutOffCallerRouteTestis now a top-level class
New analysis of commit 2ce492ff:
The doDone() propagation logic is correct — ExchangeHelper.copyResults() does not copy the interrupted flag (it's extension state), so the explicit if (subExchange.getExchangeExtension().isInterrupted()) check is necessary and correctly placed. The flag is write-once (true only, never reset), so parallel multicast with multiple recipients cannot undo an earlier propagation.
The runNotAllowed() else if branch is sound: while the branch condition only checks exchange.getException() != null, the inner logCutOffByRouteStop() call guards on instanceof RouteStoppingException before logging or calling setInterrupted(true). A business exception (NPE, etc.) reaching a stopped error handler causes logCutOffByRouteStop() to return false, leaving the exchange untouched. No DLC suppression for genuine failures.
One minor comment-accuracy nit below.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
…eStoppingException
…ndler fires When a split() is nested inside a multicast(), and the route is stopped while an exchange is inflight, two WARN messages were logged instead of one. Root cause: after the split's error handler marks the sub-exchange (.0003) as interrupted and logs WARN #1, the split's taskCallback returns control to the multicast's error handler for the outer sub-exchange (.0002). At that point .0002 has no exception yet, so the multicast's error handler sets a new RouteStoppingException on .0002 and then logs WARN #2 (interrupted=false on .0002). The existing fix in doDone() that propagates interrupted from subExchange to original only fires after the taskCallback returns, i.e. after the outer error handler's callback has already been scheduled. The outer error handler then runs with interrupted=false. Fix: in the taskCallback of MulticastReactiveTask, propagate the interrupted flag from the sub-exchange to the original exchange immediately, before exchangeResult.accept() is called. This ensures that when the outer error handler's runNotAllowed() executes, it sees interrupted=true on the original exchange and skips logging.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on davsclaus's Oct 25 18:08 review — all three findings are addressed in the two new commits (88f9508b, 5740eb32).
davsclaus findings resolved:
- ✅ Split/multicast double-WARN (finding 1, blocking):
5740eb32adds interrupted-flag propagation inMulticastReactiveTask.taskCallback— immediately, beforeexchangeResult.accept()fires. This is the correct insertion point:doDone()propagates only after all sub-exchanges complete and aggregate, which is too late for the outer error handler'srunNotAllowed(). The taskCallback path covers the reactive happy path;doDone()covers thestopOnExceptionearly-return path. Together they close the gap. - ✅ Comment inaccuracy (finding 2, nit):
88f9508breplaces "already has aRouteStoppingException" with "already has an exception" and explicitly notes thatlogCutOffByRouteStopguards oninstanceof RouteStoppingExceptionand is a no-op for other exceptions. Accurate now. - ✅ Nested test channel-count wording (finding 3, nit): Was already corrected in
2ce492ff— the class-level javadoc now says "two stopped error-handler channels in route B … plus route A's running channel."
New analysis of 5740eb32:
The fix is structurally sound. The propagation block:
if (exchange.getExchangeExtension().isInterrupted()) {
original.getExchangeExtension().setInterrupted(true);
}
exchangeResult.accept(exchange);fires in the same thread that runs the reactive callback, so no race between the write and the outer handler's read. The flag is effectively write-once (true only, never reset), so concurrent multicast recipients setting it redundantly is harmless. The existing doDone() propagation remains correct for the stopOnException early-return branch (:970-972).
RouteStopCutOffLogNestedTest.multicastSplitCutOffIsOneWarnLine directly validates the fix: a single-item split inside a multicast, latch-stopped, asserts exactly one WARN.
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Re-review of 988bdcd80e..5740eb3261. Thanks Guillaume. There is no merge of main in this range. The three new commits copy the interrupted flag from a sub-exchange to the original in MulticastProcessor, fix the runNotAllowed() comment, add a multicast+split test, and fix the test javadoc. The doStart/doStop part I approved earlier has not changed.
Status of my last review (at 988bdcd80e):
- 1. Multicast+split single WARN: done.
RouteStopCutOffLogNestedTest.multicastSplitCutOffIsOneWarnLinenow covers it, and it passes in CI with thetaskCallbackchange. ThedoDone-only version at88f9508bc1still logged two WARNs (run 40221799087), so the move intotaskCallbackwas needed. But the change breaks an existing test on main, see below. - 2. Nit, caller-test javadoc: fixed. It now says the
.process()handler sets the exception and the next outer stopped handler logs and marks the exchange. - 3. Nit, "three nested stopped channels": fixed. The text now says two stopped channels plus route A's running channel.
[Blocking] A cut-off in a direct producer inside a split no longer logs its WARN, and RouteStopCutOffLogTest fails
CI build (21, false) fails on this head in RouteStopCutOffLogTest.theCutOffIsOneWarnLine:78. The ConditionTimeout happens in Run 1, 2 and 3 (run 40225579339). This test is on main (CAMEL-25484/25502) and the PR does not touch it. It passed on 988bdcd80e. The await at :78 waits for any message at all, so the cut-off now logs nothing.
Here is the trace with the route seda:start → split(body()) → log → to("direct:shipment"), where direct:shipment has no consumer:
- On the forced stop, the seda thread is interrupted.
DirectProducer(DirectProducer.java:102-118) sets aRouteStoppingExceptionon the split item and callssetInterrupted(true)without logging the WARN. It logs only its own INFO line. - The item's handlers do not log.
SimpleTask.run()either goes torunInterrupted()(RedeliveryErrorHandler.java:846-857, no log) or, if already stopped, torunNotAllowed(). There the new branch is skipped because of!isInterrupted()(:863). - New in
5740eb3261:MulticastReactiveTask.taskCallbackcopiesinterrupted=trueto the original (MulticastProcessor.java:713-715). - The split's own channel handler is the one that logged the single WARN on main and on
988bdcd80e. It did so throughhandlePreviousFailure()→logFailedDelivery()→logCutOffByRouteStop()(:999) when it was running, or through the newrunNotAllowed()branch when it was stopped. It now seesinterrupted=trueand takesrunInterrupted()or skips therunNotAllowed()branch. So nothing is logged.
The root cause is that interrupted is now used for two things: "stop routing / skip the DLC" and "the cut-off WARN was already logged". The comment at RedeliveryErrorHandler.java:866-868 says "interrupted exchanges are already logged through their own path", but that is not true for DirectProducer. It is also not true for DefaultAsyncProcessorAwaitManager.interrupt() (:200-201) or for setException(InterruptedException) (AbstractExchange.java:613-615). Those set the flag without logging the WARN. Before this commit the flag never crossed the split boundary, so the parent handler logged the WARN.
Suggested direction (I traced it but have not run it): keep a separate marker for "cut-off WARN logged", and copy interrupted to the original only when that marker is set on the sub-exchange. The marker is set where runNotAllowed() logs and marks (:870-872). Then:
- The stopped-handler case in the nested and caller tests still copies the flag, so it logs one WARN and the caller skips its DLC.
- The
DirectProducercase copies nothing. The split's handler logs the one WARN as on main, andRouteStopCutOffLogTestpasses again. - It also fixes the scope problem in the next finding.
For example (the marker name is only illustrative):
// RedeliveryErrorHandler.runNotAllowed()
if (logCutOffByRouteStop(exchange, exchange.getException(), logger.getLevel())) {
exchange.setProperty(CUT_OFF_LOGGED, true);
exchange.getExchangeExtension().setInterrupted(true);
}
// MulticastProcessor taskCallback (and doDone / the transacted task)
if (exchange.getExchangeExtension().isInterrupted() && exchange.getProperty(CUT_OFF_LOGGED) != null) {
original.getExchangeExtension().setInterrupted(true);
original.setProperty(CUT_OFF_LOGGED, true);
}RouteStopCutOffLogTest, RouteStopCutOffLogNestedTest and RouteStopCutOffCallerRouteTest all need to pass together on the next push.
[Important] Any interrupted sub-exchange now turns off the parent's error handling, not only a route-stop cut-off
The new copy at MulticastProcessor.java:713-715 (and :997-999) does not check why the sub-exchange is interrupted. The upgrade guide (camel-4x-upgrade-guide-4_24.adoc, "Forced route-stop cut-off marks exchange as interrupted in any nested EIP") describes only the forced route stop. A scenario where this matters:
- A route has
onException(InterruptedException.class)or a dead letter channel, and a processor in a split item throwsInterruptedException.AbstractExchange.setExceptionmarks the item as interrupted (:613-615). - On main, the item's handler stops routing the item, and then the split's channel handler handles the failure on the original.
onExceptionmatches and the DLC runs. The comment inDirectProducer.java:108-109relies on this ("onException(InterruptedException.class) still matches"). - On this head, the original is interrupted. The split's handler runs
runInterrupted()(orisDone()→truewith a DLC,:302-311), so there is noonException, no DLC and no log line. - If a custom aggregation strategy clears the item's exception (a common "collect the failures" pattern), the original has no exception but
routeStop=true. The steps after the split are then skipped silently.
The same happens when an exchange blocked in a synchronous multicast (awaitManager.process, :734) is interrupted through JMX. The marker check in the previous finding limits the copy to real cut-offs. Otherwise, the guard needs at least exchange.getException() instanceof RouteStoppingException.
[Important, non-blocking] MulticastTransactedTask does not copy the flag
The copy was added only to MulticastReactiveTask's taskCallback. MulticastTransactedTask.doRun() (MulticastProcessor.java:850-856) accepts and aggregates without it. Its doDone call on the normal path also passes result.get(), the aggregated exchange, not the item. So in a transacted route, a split inside a multicast still logs one WARN per copy boundary, and a calling route still runs its DLC. The upgrade guide says "in any nested EIP". Please add the same (marker-guarded) copy there, or narrow the sentence.
[Nit] Comments credit doDone() with the propagation
In the reactive path, doDone gets result.get() (MulticastProcessor.java:516, :538), and for the default split that is the original itself. So the block at :992-999 takes effect only on the stopOnException path (:703) and the timeout path. That is why 88f9508bc1 still logged two WARNs. The comment at :992-996 and the test comments still say MulticastProcessor.doDone() propagates the flag:
RouteStopCutOffLogNestedTest.java:53,:142,:169,:233RouteStopCutOffCallerRouteTest.java:147
Pointing them at the taskCallback (and saying doDone covers stopOnException) would save the next reader the same detour.
Checklist
- Tests: the new
multicastSplitCutOffIsOneWarnLinefollows the latch →stopRoute→ release pattern, waits on the future before counting, and fails with<2>without thetaskCallbackchange (the CI run at88f9508bc1showed this). NoThread.sleep, JUnit style as in the module. The regression is inRouteStopCutOffLogTestfrom main, see the blocking finding. - Docs / upgrade guide / catalog: the upgrade-guide entry is there. Its scope ("forced route stop") is narrower than what the code does now (any interrupted sub-exchange, reactive path only). No catalog impact.
- Commit: the commit titles use the
fix:/test:prefixes; the squash titleCAMEL-25823: ...is fine. - Generated files / API: no API change. A marker property, if added, should stay internal (not a new public
Exchangeconstant without@since). - Security: no concern. The change only affects logging and error-handler routing of exchanges already cut off.
- CI:
build (21, false)fails on5740eb3261:RouteStopCutOffLogTest.theCutOffIsOneWarnLine3/3 attempts, caused by this PR (passed on988bdcd80e, test untouched).FileConsumerIdempotentKeyNameAndSizeTestandFileConsumerIncludeNameTestflaked once and passed on rerun; they are unrelated file-consumer tests. The other checks pass.
Reviewed with Claude Code on behalf of @davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
…otAllowed WARN; redesign RouteStopCutOffLogTest to use latch pattern The runNotAllowed() WARN guard used !isInterrupted() to prevent duplicate warnings from nested stopped error handlers. However, DirectProducer (CAMEL-25502) sets isInterrupted(true) before the split's error handler runs, silently suppressing the WARN and causing RouteStopCutOffLogTest to time out in CI. Fix: gate on CUTOFF_WARN_LOGGED exchange property instead. This property is set only when runNotAllowed() itself emits a WARN, not by DirectProducer, so the first WARN is always emitted while duplicates from nested handlers are still suppressed. Also redesign RouteStopCutOffLogTest to use the same latch pattern as the other tests in this PR (direct: consumer + CountDownLatch), which ensures the processing thread is not interrupted by the seda executor shutdown and the WARN is emitted deterministically after stopRoute() returns.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Re-review of 5740eb3261..c3354fd41c. Thanks Guillaume. The CUTOFF_WARN_LOGGED marker is the right direction. Gating both the runNotAllowed() WARN and the copy in taskCallback/doDone on it (RedeliveryErrorHandler.java:871-881, MulticastProcessor.java:715-718, :1000-1003) addresses my second point: an InterruptedException in a split item, or a JMX interrupt, no longer turns off the parent's onException/DLC. By reading the code, it also brings the DirectProducer cut-off back to exactly one WARN.
Still open:
RouteStopCutOffLogTestis main's test (CAMEL-25484/CAMEL-25502), and this commit rewrites it (RouteStopCutOffLogTest.java:66-172). Theseda+direct:shipmentwait was the scenario that regressed, and the new version avoids the interrupt on purpose. As a result, no test covers the DirectProducer cut-off WARN or the error registry for it any more. Could we keep main's version unchanged and add the latch scenario as a new test method? Then the original test proves the fix.- Conflicts with main, and no CI on this head.
RedeliveryErrorHandler.java(main addedCONTINUED_EXCEPTIONnext to the new constant) and the 4.24 upgrade guide conflict. Both need a "keep both" resolution. Main has changedRedeliveryErrorHandlerand Split a lot since the merge base (CAMEL-26389, CAMEL-26430, CAMEL-26445, CAMEL-26332, CAMEL-26338, CAMEL-26476). After the merge,RouteStopCutOffLogTest,RouteStopCutOffLogNestedTest,RouteStopCutOffCallerRouteTestandDefaultSupervisingRouteControllerMulticastSplitTestall need to pass together. MulticastTransactedTaskstill lacks the copy (MulticastProcessor.java:850-858). Either add the same marker-guarded block there, or narrow "in any nested EIP" in the upgrade guide.- Nits:
- The test comments still credit
MulticastProcessor.doDone()with the propagation (RouteStopCutOffLogNestedTest.java:53,:142,:169,:233;RouteStopCutOffCallerRouteTest.java:147). RouteStopCutOffLogTest.java:101uses a fully qualifiedorg.junit.jupiter.api.Assertions.fail; please import it.- The shutdown timeout went from 1 s to 5 s in the three methods, so each
stopRoutenow waits 5 s. - Main's
CONTINUED_EXCEPTION("CamelErrorHandlerContinuedException", with an "internal on purpose" comment) sits next to the new constant. A matching name (for exampleCamelErrorHandlerCutOffLogged) and the same note would keep the two consistent.
- The test comments still credit
- The PR description still describes an
isInterruptedgate and "three changes in MulticastProcessor". Could it mention the marker and the propagation, the second method in the nested test, and theRouteStopCutOffLogTestchange?
The doStart/doStop lifecycle part I approved earlier is unchanged. All 10 earlier review threads are resolved.
Reviewed with Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review and davsclaus's Nov 3 review on c3354fd41c — the CUTOFF_WARN_LOGGED marker approach is correct, and the taskCallback + doDone() dual-path propagation closes the double-WARN race. Two blockers from the prior review remain unaddressed.
What's resolved in c3354fd41c:
- ✅
CUTOFF_WARN_LOGGEDmarker replacesisInterrupted()gate inrunNotAllowed()— correctly avoids suppressing the first WARN whenDirectProducerpre-setsinterrupted=true - ✅ Reactive path (
taskCallback) propagation updated to gate onCUTOFF_WARN_LOGGEDinstead ofisInterrupted() - ✅
doDone()propagation likewise updated to gate onCUTOFF_WARN_LOGGED - ✅ Merge with upstream/main (
33a1defd) — merge conflicts resolved
Still open:
MulticastTransactedTask.doRun()still lacks theCUTOFF_WARN_LOGGEDpropagation — in a transacted route, a nested split inside a multicast still logs one WARN per error-handler boundary (see inline comment)RouteStopCutOffLogTestwas fully rewritten — the originalseda + direct:shipmentscenario (DirectProducer cut-off, CAMEL-25502) is no longer covered; davsclaus explicitly asked to keep it and add the latch scenario as a second method (see inline comment)
This review was generated by an AI agent, Hermès on behalf of @gnodet.
…nits - RouteStopCutOffLogTest: restore main's original seda:start → split → direct:shipment scenario (3 test methods) alongside the new latch-based scenario (3 processedCutOff* methods) in the same class with both routes in createRouteBuilder() - MulticastTransactedTask.doRun(): add CUTOFF_WARN_LOGGED propagation before doDone(), consistent with the reactive path in taskCallback - Rename CUTOFF_WARN_LOGGED value to CamelErrorHandlerCutOffLogged to match CONTINUED_EXCEPTION naming; add 'internal on purpose' comment - Update test comments: MulticastProcessor.doDone() → MulticastProcessor since propagation now happens in both taskCallback and doRun() - Use static import for fail() in RouteStopCutOffLogTest
|
Following up on davsclaus's Nov 3 review — all four items are addressed. What's addressed:
This review was generated by an AI agent, Hermès on behalf of @gnodet. |
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Following up on davsclaus's Nov 3 review and my earlier comment — all open blockers and nits are addressed in commit fbedbc0366ea.
What's resolved:
- ✅
MulticastTransactedTask.doRun()missingCUTOFF_WARN_LOGGEDpropagation (blocking): the same marker-guarded block (if (exchange.getProperty(CUTOFF_WARN_LOGGED, ...) { original.setInterrupted(true); original.setProperty(CUTOFF_WARN_LOGGED, TRUE); }) is now added beforeexchangeResult.accept()inMulticastTransactedTask.doRun(), consistent with the reactive path intaskCallback. The upgrade guide's "in any nested EIP" claim now holds for the transacted path too. - ✅
RouteStopCutOffLogTestoriginal scenario restored (blocking): main's three test methods (theCutOffIsOneWarnLine,theCutOffIsNotInTheErrorRegistry,theCutOffIsInTheErrorRegistryWhenIncluded) are restored unchanged in the same class, with the originalseda:start → split → direct:shipmentroute increateRouteBuilder(). The three new latch-based methods (processedCutOff*) use a separateslow-latchroute ID to avoid interference. Both the DirectProducer cut-off path and the stopped-error-handler-chain path are now covered. - ✅ Comment nits —
doDone()→MulticastProcessor: all five occurrences (RouteStopCutOffLogNestedTest.java:53,:142,:169,:233;RouteStopCutOffCallerRouteTest.java:147) corrected. - ✅
failstatic import inRouteStopCutOffLogTest:import static org.junit.jupiter.api.Assertions.failadded. - ✅
CUTOFF_WARN_LOGGEDconstant renamed: value changed fromCamelRedeliveryErrorHandlerCutOffWarnLoggedtoCamelErrorHandlerCutOffLoggedto matchCONTINUED_EXCEPTION's naming; "internal on purpose" comment added, consistent with the neighbour. - ✅ PR description updated: now describes the
CUTOFF_WARN_LOGGEDmarker, the three propagation insertion points (taskCallback,doRun(),doDone()), the restored + new latch-basedRouteStopCutOffLogTestmethods, and theMulticastTransactedTaskcoverage.
davsclaus nit 4 — shutdown timeout: the original three seda + direct:shipment tests use 1 s (as on main); the new latch-based tests keep 5 s (required for the stopRoute to force-stop with the shutdown timeout). No change needed.
No new issues found.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on davsclaus's Nov 3 review — all 3 blocking findings are addressed in commit fbedbc03.
Blockers resolved:
- ✅
RouteStopCutOffLogTestfully rewritten (blocking 1): original 3 methods (theCutOffIsOneWarnLine,theCutOffIsNotInTheErrorRegistry,theCutOffIsInTheErrorRegistryWhenIncluded) are untouched. 3 new latch-based methods (processedCutOff*) added under a separateslow-latchroute to avoid interference. Both the DirectProducer cut-off path and the stopped-error-handler-chain path are now covered. - ✅
MulticastTransactedTask.doRun()propagation (blocking 2): the sameCUTOFF_WARN_LOGGED-gated block is now added beforedoDone()indoRun(), matching the structure intaskCallback. The upgrade guide's "in any nested EIP" claim now holds for the transacted path too. - ✅ Conflicts with main (blocking 3): diff applies cleanly.
Important finding resolved:
- ✅ Over-propagation of
interruptedflag (important 4): theCUTOFF_WARN_LOGGEDgate fully addresses this. Propagation only fires when a stopped error handler already logged the WARN.DirectProducersetsinterruptedwithout settingCUTOFF_WARN_LOGGED, so its interruption is correctly NOT forwarded to the original — the outer handler still logs its own WARN.
Nits:
- ✅
failis properly statically imported in all test files - ✅ Original 3 tests keep
setTimeout(1), new latch-based tests usesetTimeout(5)— exactly as requested - ✅ Constant named
CamelErrorHandlerCutOffLoggedwith "Internal on purpose" comment, consistent withCONTINUED_EXCEPTION - ✅ Test comments now correctly point to
MulticastProcessor.taskCallback/doRun()as the propagation sites (notdoDone())
One observation on doStop()/doShutdown(): the change from passing errorHandlers (the Map) directly to errorHandlers.values() is a latent fix beyond the stated scope — previously, ServiceHelper.stopService(Object...) received the Map itself, which is not a Service or Iterable<Service>, making it a silent no-op. The new code properly stops the individual processors. This is correct and welcome; worth a mention in the squash commit message so the change is traceable.
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
Summary
Defer error handler startup in
MulticastProcessorfromdoInit()todoStart()to fix compatibility withSupervisingRouteControllerwhen a producer cannot connect at context startup.SplitterextendsMulticastProcessorand is fixed by this change as well.Root Cause
MulticastProcessor.doInit()callswrapInErrorHandler()for each child processor, which eagerly invokedServiceHelper.startService(answer)on the wrapped error handler. Starting the error handler starts the wrappedSendProcessor, which in turn starts the underlyingProducer(viacamelContext.addService(producer, true, true)).This producer startup happens:
CamelContextis fully startedSupervisingRouteControllerretry scopeA producer that cannot connect yet (e.g. broker is temporarily down) propagates the exception directly to
context.start(), failing the entire context — even though the route was marked for supervision.Fix
Three changes in
MulticastProcessor:In
wrapInErrorHandler(): replaced the unconditionalServiceHelper.startService(answer)with a conditional start gated onisRunAllowed().isRunAllowed()returnstrueinSTARTING,STARTED,SUSPENDING, andSUSPENDEDstates, covering both wrappers built at runtime (e.g. byRecipientListProcessor) and those built during route startup. DuringdoInit()the processor is inINITIALIZINGstate, soisRunAllowed()isfalseand the built error handler is cached inerrorHandlersbut not started. The wrapper is now started before being published to the cache, avoiding a race where a concurrent exchange could take the cache-hit path and use an unstarted wrapper.In
doStart(): addedServiceHelper.startService(errorHandlers.values())to start any pre-built error handlers that were deferred fromdoInit(). This runs under the route lifecycle managed bySupervisingRouteController, so failures are caught and the route is scheduled for retry with back-off.ServiceHelper.start()is idempotent (no-op if already started), so there is no double-start risk when new error handlers are built at runtime anddoStart()iterates overerrorHandlers.values().In
doStop()anddoShutdown(): correctedServiceHelper.stopService(errorHandlers)/ServiceHelper.shutdownService(errorHandlers)to passerrorHandlers.values()instead of the rawMap. The previous code silently did nothing becauseMapdoes not implementIterable; now the cached per-branch error handlers are correctly stopped and shut down with the route.One additional fix in
RedeliveryErrorHandler.SimpleTask.runNotAllowed():CUTOFF_WARN_LOGGEDmarker and single-WARN guarantee: when an exchange already carries aRouteStoppingExceptionand its error handler is stopped at route shutdown,runNotAllowed()now guardslogger != nullbefore logging (matching the guard inlogFailedDelivery, sinceLegacyDefaultErrorHandlerReifierpasses a null logger). It setsCamelErrorHandlerCutOffLoggedon the exchange after logging the cut-off WARN, so nested stopped error handlers (e.g. a split inside a multicast) do not each emit their own WARN for the same exchange. The marker also triggersexchange.getExchangeExtension().setInterrupted(true), which causes any caller route that passed the same exchange into the stopped route viadirectto skip redelivery,onException, and the dead letter channel — the same behavior asDirectProducer(CAMEL-25502). The marker (notisInterrupted()) is used as the gate, so a pre-existing interrupt fromDirectProduceror a JMX interrupt does not suppress the WARN.The marker is propagated from sub-exchange to original in:
MulticastReactiveTask.taskCallback— the normal reactive path (beforeexchangeResult.accept())MulticastTransactedTask.doRun()— the transacted path (beforeexchangeResult.accept())MulticastProcessor.doDone()— thestopOnExceptionearly-return pathUpgrade Guide
An upgrade guide entry has been added to
camel-4x-upgrade-guide-4_24.adocdocumenting the behavioral changes for users who may have been relying on the previous (incorrect) lifecycle behavior.Tests
New test class
DefaultSupervisingRouteControllerMulticastSplitTestwith two test methods:testSplitWithFlakyProducer()— asplit().to("flaky:target")route where the producer fails its first 2doStart()calls. Assertscontext.start()does not throw and the route eventually reachesStartedafter the supervising controller retries.testMulticastWithFlakyProducer()— same scenario withmulticast().to("flaky:target").Updated test class
RouteStopCutOffLogTest(CAMEL-25484/25502):seda:start → split → direct:shipment, DirectProducer cut-off):theCutOffIsOneWarnLine(),theCutOffIsNotInTheErrorRegistry(),theCutOffIsInTheErrorRegistryWhenIncluded(). These test the pre-existing CAMEL-25502 fix whereDirectProducersets the interrupt; theCUTOFF_WARN_LOGGEDmarker ensures exactly one WARN is logged even in that case.processedCutOffIsOneWarnLine(),processedCutOffIsNotInTheErrorRegistry(),processedCutOffIsInTheErrorRegistryWhenIncluded(). These test the new path whererunNotAllowed()in a stopped split/filter/multicast channel logs the WARN.New test class
RouteStopCutOffLogNestedTestwith two test methods:nestedEipCutOffIsOneWarnLine()— afilterwrapping a blocked processor, where route A stays running. Route B is stopped; the test asserts exactly one WARN for the exchange.multicastSplitCutOffIsOneWarnLine()— asplitnested inside amulticast, latch-stopped. Asserts exactly one WARN, validating that theCUTOFF_WARN_LOGGEDpropagation intaskCallbackprevents a second WARN on the original exchange.New test class
RouteStopCutOffCallerRouteTestwith one test method:callerRouteDlcNotCalledOnNestedEipCutOff()— route A (with a dead letter channel) sends viadirectto route B, which is amulticast()wrapping asplit(). Route B is force-stopped while an inflight exchange is blocked inside the nested EIP. The test asserts that route A's dead letter channel is NOT called: thesetInterrupted(true)marker set on the exchange by route B's error handler propagates back to route A, which then skips redelivery,onException, and the DLC.Fixes #CAMEL-25823
Hermes Agent (Claude Sonnet 4.6) on behalf of Guillaume Nodet