Repository navigation
CAMEL-26286: camel-disruptor - removing one of the routes that consume from a disruptor endpoint no longer shuts down its disruptor - #28639
Conversation
…e from a disruptor endpoint no longer shuts down its disruptor Every route that consumes from the same disruptor uri shares one endpoint, and removing a route stops and shuts down the endpoint of its consumer. That shut down the disruptor of the other routes: they received nothing any more and producers failed with "Disruptor was not started". The endpoint now ignores stop and shutdown while other consumers use it, as SedaEndpoint does. The consumer of a suspended route counts too: a suspended consumer leaves the disruptor (as before) but is kept as a suspended consumer of the endpoint until it resumes or stops, so that removing another route does not shut down the disruptor that the suspended routes resume on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Mirrors the SedaEndpoint stop/shutdown guard. Counting suspended consumers is right (doStop from SUSPENDED removes it, and the lock is reentrant), not counting producers is justified and covered by a control test, and context shutdown still stops the endpoint because the consumers stop first. Will approve once CI is green.
Nit: the emptiness check at DisruptorEndpoint.java:289/299 runs outside the service lock, so a consumer starting concurrently could race it (low risk).
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.
|
🧪 CI tested the following changed modules:
🔁 1 test passed only after a retry on JDK 21 (1 retried attempt) Recovered flaky tests on JDK 21 (1)
🔬 Scalpel shadow comparison — Scalpel: 8 of 709 tested, 27 compile-only — current: 8 all testedMaveniverse Scalpel detected 8 affected modules (current approach: 8). Skip-tests mode would test 8 modules (1 direct + 7 downstream), skip tests for 27 (generated code, meta-modules) Modules Scalpel would test (8)
Modules with tests skipped (27)
All tested modules (35 modules, 5m 20s total)Total reactor time: 5m 20s
Top 20 slowest modules:
|
…r its lock before it stops or shuts down DisruptorEndpoint.stop() and shutdown() checked that no consumer (and no suspended consumer) uses the endpoint before they took the service lock, so a consumer that registered between the check and the stop (onStarted takes the same lock) ran on an endpoint that then stopped its disruptor. The check now runs under the lock; the lock is reentrant, so super.stop() and super.shutdown() take it again as before. DisruptorRemoveRouteOfSharedEndpointTest holds the lock as a registering consumer does, asks the endpoint to stop from another thread, registers a consumer and then releases the lock: the endpoint must stay started. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
@davsclaus thanks for the review. The nit is fixed in dca9d85 (one new commit on top): Test: No open follow-ups. Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
LGTM. The fix is correct and mirrors the established SedaEndpoint guard pattern exactly.
Review summary:
- Lock reentrancy (
onSuspended→onStoppedboth acquirelock): safe —BaseService.lockis aReentrantLock. - Consumer lifecycle tracking: leak-free —
onStopped,onStarted(resume), andonSuspendedall maintainsuspendedConsumersconsistently. stop()/shutdown()override: correct — endpoint staysSTARTEDwhile active or suspended consumers exist; proceeds normally when the last consumer leaves.- Lock ordering (endpoint lock → DisruptorReference lock): no inversion, no deadlock risk.
- Test coverage: four targeted scenarios including a lock-hold race test (
consumerRegisteringWhileTheEndpointIsAskedToStopKeepsTheEndpointStarted) that exercises the TOCTOU window directly. Author confirmed three of four tests fail onmainwithout the fix.
Backport note: The PR description targets camel-4.22.x, 4.18.x, and 4.14.x, but no port/* labels are set. Please add port/camel-4.22.x, port/camel-4.18.x, and port/camel-4.14.x to trigger the backport bot.
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.
The changes since my review address my points; CI is green. Thanks.
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.
Description
CAMEL-26286
Every route that consumes from the same
disruptor:uri shares oneDisruptorEndpoint. When a route is removed, the core stops and shuts down the endpoint of its consumer (RouteService.doShutdown), also when other routes still consume from it.DisruptorEndpoint.doStopthen callsDisruptorReference.removeEndpoint, which shuts down the disruptor and its executor, as this endpoint is the only one of its reference (another endpoint shares the reference only for a uri with the same queue name and other options). So with three routesfrom("disruptor:orders?multipleConsumers=true"), removing one of them (a route template removed, a reload, JMX or controlbus) leaves the two others without messages, and every send fails withIllegalStateException: Disruptor was not started(caused byDisruptorNotStartedException). ACamelContext.stop()afterwards takes about 10 seconds. camel-seda has the same sharing and guards against it inSedaEndpoint.stop()/shutdown(); the same scenario onseda:keeps delivering to the other two routes.Changes:
DisruptorEndpointoverridesstop()andshutdown(): they do nothing (DEBUG log) while other consumers still use the endpoint, so the disruptor stays up for them. Without consumers, the endpoint stops and shuts down as before. The check runs under the endpoint's service lock, which a consumer takes to register, so a consumer that starts at the same time is either seen by the check or registers after the stop (review follow-up).DisruptorConsumerleaves the disruptor (as before, so the disruptor stops handing it events), but the endpoint now keeps it in a set of suspended consumers until it resumes or stops. Otherwise suspending two routes and removing a third left the endpoint without consumers, so it was stopped and shut down, and the suspended routes resumed on a disruptor without them: they never received anything again.SedaEndpoint, producers are not counted. A producer of the uri of a removed route, such as the cached producer of aProducerTemplate, would otherwise keep the old endpoint in the disruptor, and the endpoint that replaces it (the context removes the endpoint from its registry when no other route uses it) could not join it:DisruptorReferencekeeps its endpoints in a set based onequals, that is on the uri. The first version of this change counted producers too, andDisruptorRemoveRouteThenAddAgainTestfailed with it.port/*labels): camel-4.22.x, 4.18.x and 4.14.x have the sameDisruptorEndpoint.doStop(read, not run).Tests:
DisruptorRemoveRouteOfSharedEndpointTest, MockEndpoint expectations as in the module's other tests:removingOneOfSeveralConsumersKeepsTheOthersConsumingfails on main (3 runs):IllegalStateException: Disruptor was not started.removingAConsumerRouteKeepsTheSuspendedOnes: routes a and c suspended, route b removed, a and c resumed, one message: without the suspended consumers it fails after the mock's wait withmock://c Received message count. Expected: <1> but was: <0>.consumerRouteRemovedAndAddedAgainWhileAProducerRouteUsesTheQueue(the consumer route reloaded while a producer route stays) passes on main and with the change; it is the control for the path the change leaves as it is.consumerRegisteringWhileTheEndpointIsAskedToStopKeepsTheEndpointStarted(review follow-up): the test holds the lock as a registering consumer does, another thread asks the endpoint to stop and waits for the lock, then a consumer registers. With the check outside the lock the endpoint stops (expected: <true> but was: <false>). No sleep: it waits until the stopping thread is queued on the lock.DisruptorRemoveRouteOfSharedEndpointTestpassed 20 runs in one JVM.How it was found: a random lifecycle-sequence harness (stop, start, suspend, resume, remove and add of several consumer routes on shared endpoints, with a probe message after every step) over seda, disruptor, timer, scheduler and other consumers shrank the failure to a single remove.
Not changed: producers are not counted (above), so removing the last consumer route of a uri still stops its endpoint while a producer of another route sends to it, as on main.
Also not changed: the core stops the endpoint of a removed route's consumer even when other routes still use it. Other components whose endpoint
doStopreleases something the other routes of the same uri still use may have the same problem; I can file a follow-up JIRA for the ones I can reproduce.Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.(I built camel-disruptor and ran its tests, including the formatter and import-sort plugins. No generated file changes. I did not run the full root build.)
AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a
Co-Authored-Bytrailer.Claude Code on behalf of allthingssecurity
🤖 Generated with Claude Code