Repository navigation
Conversation
…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>
Contributor
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
Contributor
|
🧪 CI tested the following changed modules:
🔁 6 tests passed only after a retry on JDK 21 (6 retried attempts) Recovered flaky tests on JDK 21 (6)
🔬 Scalpel shadow comparison — Scalpel: 567 of 707 tested, 26 compile-only — current: 568 all testedMaveniverse Scalpel detected 567 affected modules (current approach: 568). Skip-tests mode would test 567 modules (3 direct + 565 downstream), skip tests for 26 (generated code, meta-modules) Modules only in current approach (1)
Modules Scalpel would test (567)
Modules with tests skipped (26)
Build reactor — dependencies compiled but only changed modules were tested (3 modules, 8.6s total)Total reactor time: 8.6s
Top 20 slowest modules:
|
gnodet
marked this pull request as ready for review
October 21, 2026 11:34
oscerd
requested changes
Oct 21, 2026
oscerd
left a comment
oscerd
left a comment
Contributor
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.
…rs lifecycle, improve test and upgrade guide
gnodet
changed the title
Contributor
Author
|
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
reviewed
Oct 21, 2026
gnodet-bot
left a comment
gnodet-bot
left a comment
Contributor
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
reviewed
Oct 21, 2026
gnodet-bot
left a comment
gnodet-bot
left a comment
Contributor
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
requested changes
Oct 21, 2026
davsclaus
left a comment
davsclaus
left a comment
Contributor
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
reviewed
Oct 21, 2026
gnodet-bot
left a comment
gnodet-bot
left a comment
Contributor
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
reviewed
Oct 21, 2026
gnodet-bot
left a comment
gnodet-bot
left a comment
Contributor
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
approved these changes
Oct 21, 2026
davsclaus
left a comment
davsclaus
left a comment
Contributor
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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
Two changes in
MulticastProcessor:In
wrapInErrorHandler(): replaced the unconditionalServiceHelper.startService(answer)with a conditional start gated onisRunAllowed().isRunAllowed()returnstrueinSTARTING,STARTED, 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().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").Fixes #CAMEL-25823
Hermes Agent (Claude Sonnet 4.6) on behalf of Guillaume Nodet