Repository navigation
CAMEL-25861: camel-azure-cosmosdb - the consumer starts the change feed processor again when it cannot start - #28637
Conversation
…ed processor again when it cannot start The consumer subscribed once to ChangeFeedProcessor.start() in doStart. The start reads the lease and feed containers, so it fails when Cosmos DB cannot be reached yet, the lease container does not exist yet or a role assignment is not applied yet. The failure went to the exception handler and the processor was never started again: the route stayed started and never received a change, until it was restarted. Start it again with a backoff (1 s, up to 30 s) until it starts or the consumer is stopped, logging one WARN line for each failed attempt (the stack trace at DEBUG). Each attempt after a failure uses a new change feed processor: one whose start failed while it initialized its leases throws "Partition load balancer already started" from every later start(). The attempt builds it, not the retry hook: building a change feed processor calls block() (and with the create*IfNotExists options creates the containers), and the failure of a start is signalled on a thread of the Cosmos DB client or of a Reactor scheduler, where block() throws, which ended the retry. Each attempt runs on the boundedElastic scheduler, where it may block, and a failure to build the processor is retried as a failure to start it. When the processor starts after failed attempts, one INFO line says so. On stop, the pending attempts are cancelled, and stop() is only called on a processor that has started (it throws on one that has not). The change feed processor is created in a package-private method, so that the test can give its own. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The retry logic is well-structured and the problem analysis in the PR description is thorough. One concurrency gap in doStop() needs attention before this merges.
Missing backport labels: This is a bug fix for a consumer that silently stops receiving changes when Cosmos DB is temporarily unavailable — users with transient network partitions or delayed role assignment rollouts are affected until they restart the route. Given the similar fix in camel-kubernetes (CAMEL-25857, PR #28380) was backported, this should also carry port/camel-4.22.x and port/camel-4.18.x labels.
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.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
…ose start ends after the consumer stopped doStop cancelled the attempts to start and called stop() only on a processor that isStarted(). In azure-cosmos 4.75, isStarted() is false and stop() throws until the processor has built its partition manager, so a processor whose start was before that point was left to the cancel, and was stopped only if the cancel stopped its start in time. A start that failed while it loaded the leases was not stopped either: the workers of the leases it had loaded before, which run outside the Mono that start() returns, kept running next to the next processor. The processor, the attempts to start it and doStop now hand over under a lock, with a generation that doStart and doStop increment. An attempt of an older generation does not start the processor it built, or stops it when its start ends if doStop came just as it called start(). doStop does not cancel a start that is in flight: it stops the processor if it can already (isStarted()), and the end of the start is always observed and signalled under the lock: before doStop, which then stops the processor as before, or after it, and then the attempt stops the processor if it started. A start that fails is stopped too if it started. A restart initializes the consumer again, which builds a new processor, as before; the attempt of the old generation still stops the old one. A build that fails because doStop interrupted it is not reported as a failure. The tests of CosmosDbConsumerFirstStartFailsTest deliver their change while start() runs, so the consumer may stop before the start completed: their processors now stop once, as the real one, and the tests wait for the processor to be stopped. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
@gnodet thanks for the review. Both points are addressed in d1af39f (one new commit on top), see the inline replies.
I updated the description to match. Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The follow-up commit (d1af39ff2c0b) fully addresses both findings from the previous review.
Processor leak race (thread 1): The lock+generation approach is the right solution given the SDK's constraints. The author's analysis of azure-cosmos 4.75's IncrementalChangeFeedProcessorImpl (verified via javap) explains exactly why neither of the two suggested fixes was sufficient:
- Unconditional
stop()catches the pre-partition-manager case but misses the lease-worker leak - A
volatile boolean stoppingflag would reset on restart, breaking the generation invariant
The generation counter under a final Object lock handles all six stop-while-starting scenarios, confirmed by CosmosDbConsumerStopWhileStartingTest. The acknowledged residual window (lease workers running briefly until the start ends when stop arrives before the partition manager exists) is unavoidable given stop() throwing until the partition manager is built — this is an SDK constraint, not a code defect.
onErrorListener (thread 2): The author's reasoning is correct — the handler is reachable when the Reactor scheduler rejects the delay before the next retry attempt. The comment added (// the retries are not limited...) explains this adequately. I withdraw the suggestion to remove it.
Backport labels: The author confirmed both commits apply cleanly to camel-4.22.x and camel-4.18.x (verified with git merge-tree, test class needs CamelTestSupport import adjusted on 4.18.x). Since the contributor cannot set labels, please add port/camel-4.22.x and port/camel-4.18.x.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
|
🌟 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.
gnodet-bot's two earlier findings (leak race in stop, onErrorListener) are fixed and the hand-off with lock plus generation counter holds up for stop/restart/late-callback interleavings. Will approve once CI is green.
- Question (
CosmosDbConsumer.java:114/128): a failed start no longer reaches the consumer's ExceptionHandler; it is one WARN per attempt, forever, sobridgeErrorHandlerusers lose the signal. Add a line to the upgrade guide or make it a deliberate no. - Suggestion: the consumer reports started while never connected; a health-check hook could follow.
- Nit:
createChangeFeedProcessor()reassignsclientWrapperas a side effect on boundedElastic threads (harmless, one attempt at a time). The body cites reactor-core 3.8.7, the parent pom has 3.8.8.
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 708 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 14s total)Total reactor time: 5m 14s
Top 20 slowest modules:
|
…e consumer's ExceptionHandler - The first failed attempt of each start of the consumer goes to its ExceptionHandler, as the one failure to start did before the attempts were retried, so with bridgeErrorHandler=true the error handler gets it once rather than once per attempt. Each further attempt is one WARN line, as before. A failure that ends after the consumer stopped is logged at DEBUG only. - createChangeFeedProcessor() builds the CosmosAsyncClientWrapper it uses as a local: it was a field that each attempt reassigned on a boundedElastic thread. - The 4.24 upgrade guide says that the consumer starts the change feed processor again when its start fails, and how the failures are reported. CosmosDbConsumerFirstStartFailsTest gets a test with two failed starts and a third that works: the ExceptionHandler gets the first failure only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
@davsclaus thanks for the review. Changes in 45f45d8 (one new commit on top):
camel-azure-cosmosdb: 58 tests, 0 failures (22 ITs skipped); Open follow-up: the health check you suggested (the consumer reports started while the processor is not). I would do it as a separate JIRA, a readiness check that is DOWN until the change feed processor has started. Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The change looks good. The fix correctly addresses the issue by separating the concern of starting the change feed processor from the endpoint lifecycle management.
Summary of changes:
onStartAttemptFailednow routes the first failure to the configuredExceptionHandler(instead of always retrying silently), giving users visibility into connectivity problemsclientWrapperis properly scoped as a local variable increateChangeFeedProcessor(), eliminating the unwanted side-effect on the field- The
Mono.createpattern withisStale()guard is standard Reactor idiom — whendoStop()disposes the subscription, returning early from the sink is the correct way to handle cancellation
No issues found. Backport labels (port/camel-4.22.x, port/camel-4.18.x) may be worth considering if this fix is relevant to LTS branches.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
oscerd
left a comment
oscerd
left a comment
There was a problem hiding this comment.
Read the consumer change closely — this is tricky async-lifecycle code (a start that is retried, runs on a blocking scheduler, and can complete after the consumer stops), so it deserved the scrutiny. It's very carefully done.
The fix itself is right: the change feed processor's start() reads the lease/feed containers, so it legitimately fails when Cosmos DB isn't reachable yet, the lease container doesn't exist, or a role assignment hasn't propagated — and before, that left the route up but silent forever. Retrying on Schedulers.boundedElastic() (correct, since building/starting blocks and would be rejected on a non-blocking scheduler) with Retry.backoff(…, 1s).maxBackoff(30s) until it starts or the consumer stops is the sensible behaviour.
What I checked carefully is the stop/in-flight-start race, and the locking discipline holds up:
generationis bumped underlockin bothdoStartanddoStop, and every attempt carries itsstartGeneration;isStaleis the single comparison. So oncedoStopruns, in-flight attempts are stale and won't install or start a processor.- The key invariant — the end of a
start()is observed under the lock, so eitherdoStopseesstartingand deals with the processor, or the completion handler (onStarted/onStartFailed) sees the stale generation and stops it — is sound. A processor that finishes starting after the consumer stopped is stopped byonStarted; a partial start whose lease workers already came up is stopped byonStartFailed; anddoStop, for a mid-start processor, both nulls it and callsstopIfStarteditself. No path leaks a running processor. - Double-stop is safe:
stopIfStartedgates onisStarted()and swallowsIllegalStateException, with a comment thatdoStopand the start's end can both reach it. And rebuilding a fresh processor after a failed start is the right handling of azure-cosmos marking the load balancer started before the leases initialise (Partition load balancer already started).
The error-reporting contract is preserved thoughtfully: the first failure of each start still goes to the ExceptionHandler (so bridgeErrorHandler routes it once, not once per attempt), with subsequent attempts as a single WARN line each and the stack trace at DEBUG — and a stale failure (consumer stopped) is DEBUG only, not reported. The two new tests target exactly the two behaviours that matter (…FirstStartFails = the retry, …StopWhileStarting = the race), and this PR actually has CI (re-running at the moment).
The upgrade-guide entry is accurate and complete — it documents the new retry/backoff, the old "stayed started and never received a change" behaviour, and the reporting contract.
Two non-blocking notes:
- The retry is effectively unbounded (
Long.MAX_VALUE), which matches the intent ("until it starts or the consumer stops"), but a permanently-misconfigured consumer (lease container never created, role never granted) will then emit a WARN roughly every 30s forever. That's defensible as visibility, just flagging that a stuck consumer is now a steady WARN drip rather than a single failure — acamel.*knob to cap attempts could be a future nicety, not needed here. failedStartsas avolatile longis fine here (written/read along the one Reactor chain, reset indoStartwhich happens-beforesubscribe); no concern, just noting I looked.
No blockers.
Claude Code on behalf of oscerd
|
@oscerd thanks for the close read. Agreed on both notes. The steady WARN every ~30s for a consumer that can never start is intended, so the problem stays visible. An option to cap the attempts (and then fail the route or report through the health check) fits the health-check follow-up I mentioned to Claus, and I'll file it as one follow-up JIRA rather than add it here. Claude Code on behalf of allthingssecurity |
Description
CAMEL-25861
CosmosDbConsumersubscribed once toChangeFeedProcessor.start()indoStart. The start reads the lease and feed containers, so it fails when Cosmos DB cannot be reached when the route starts, when the lease container does not exist yet (createLeaseContainerIfNotExistsisfalseby default), or when the role assignment of a managed identity is not applied yet. The failure went to the exception handler ("Error processing exchange") and nothing started the processor again: the route stayed started and never received a change until it was restarted. Stopping such a route also failed, because the SDK'sstop()throws on a processor that has not started.(With
createDatabaseIfNotExists,createContainerIfNotExistsor their lease variants, building the processor indoInitalready reads or creates the containers, so when Cosmos DB cannot be reached at that moment the route fails to start, as before; the retry covers the start that fails after that.)This is the same pattern as CAMEL-25857 in camel-kubernetes.
Changes:
doStartstarts the processor withMono.defer(() -> changeFeedProcessor.start()).retryWhen(Retry.backoff(...)): 1 s, then doubling up to 30 s, until it starts or the consumer is stopped. The first failed attempt of each start goes to the exception handler of the consumer, as the one failure did before, so withbridgeErrorHandler=trueit is routed to the error handler once and not once per attempt; each further failed attempt is logged as one WARN line with its cause, and the stack trace at DEBUG (the attempts go on every 30 s for as long as Cosmos DB cannot be reached). A failure that ends after the consumer stopped is not reported. (Review follow-up: before, no attempt went to the exception handler.)PartitionManagerImpl.start()callsPartitionLoadBalancerImpl.start()while it assembles the start, which marks the load balancer as started before the lease bootstrap runs, so after a failure in the bootstrap (a 429 or 503 while the leases are created, no write permission on the lease container) every laterstart()throwsIllegalStateException: Partition load balancer already started. A failure before that (the feed or lease container cannot be read) could be retried on the same processor; a new one works in both cases.boundedElasticscheduler (subscribeOnbeforeretryWhen, so each resubscription goes through it). Building a processor blocks:CosmosDbContainerOperations.captureEventsWithChangeFeedzips the feed and lease containers and callsblock()(with thecreate*IfNotExistsoptions, it creates them first). A failure to build the processor is retried like a failure to start it. The failure of a start is signalled on a thread of the Cosmos DB client or of a Reactor scheduler, whereblock()throwsIllegalStateException(BlockingSingleSubscriber.blockingGetchecksSchedulers.isInNonBlockingThread()first, in reactor-core 3.7.11, which camel-azure-cosmosdb gets from azure-core 1.57.0, as in 3.8.x; the parent'sreactor-version3.8.8 does not manage reactor-core); thrown from the retry hook, that would end the retry, and the consumer would again stay started without receiving anything. When the processor starts after failed attempts, one INFO line says so.createChangeFeedProcessor(), which the test overrides (instead of setting the private field by reflection). It builds the client wrapper it uses as a local, no longer as a field that each attempt reassigned on aboundedElasticthread (review follow-up).doStopcancels the pending attempts and stops the processor. Astart()that is in flight is not cancelled: in azure-cosmos 4.75,isStarted()is false andstop()throws (The ChangeFeedProcessor instance has not fully started) until the processor has built its partition manager, and while it loads the leases it starts a worker for each lease this host owns, outside theMonothatstart()returns. The processor, the attempts anddoStophand over under a lock, with a generation thatdoStartanddoStopincrement: a processor built after the stop is not started,doStopstops a starting processor if it can already, and the attempt observes the end of every start and stops the processor then if the consumer stopped meanwhile. A start that fails is stopped too if it started (a failure while loading the leases leaves the workers of the leases loaded before running). (Review follow-up.)No option changed. The 4.24 upgrade guide says that the consumer starts the change feed processor again when its start fails, and how the failures are reported (review follow-up).
Tests:
CosmosDbConsumerFirstStartFailsTest(5 tests).consumerStartsTheChangeFeedAgainWhenTheFirstStartFails: a change feed processor whose firststart()fails and whose second start delivers one change. Without the change it fails after 20 s withthe change feed processor was started 1 time(s), the consumer is started but receives nothing. With the change the change is received about 1 s after the WARN.consumerStartsANewChangeFeedProcessorWhenTheFirstOneFailedWhileInitializingTheLeases(same class): the first processor fails its start and then throwsPartition load balancer already startedas the real one does; the consumer must start a second processor. Retrying the same processor fails after 20 s (the failed change feed processor was started 5 time(s), no new one was started).consumerStartsANewChangeFeedProcessorWhenTheStartFailsOnAReactorThread(same class): the first processor's start fails on a Reactorparallelthread (Mono.delay(...).then(Mono.error(...))), and the test'screateChangeFeedProcessor()ends inMono.zip(...).block()as the real one does. With a retry hook that builds the processor (the first version of this change) the hook runs onparallel-2,block()throws and the retry ends: it fails after 20 s (no new change feed processor was started, built on the threads [main, parallel-2]).theFirstFailedStartGoesToTheExceptionHandlerAndTheFollowingOnesAreLogged(same class, review follow-up): two failed starts, then the third starts; the consumer's exception handler gets the first failure only. Without the follow-up it gets none.consumerTriesAgainWhenANewChangeFeedProcessorCannotBeBuilt(same class): after the first start fails, building the next processor throws once (as withcreateLeaseContainerIfNotExists=truewhile Cosmos DB cannot be reached), and the third one starts. With the first version of this change, the exception from the retry hook ended the retry: it fails after 20 s.CosmosDbConsumerStopWhileStartingTest(6 tests), with a processor that behaves as the azure-cosmos 4.75 one (isStarted()once its partition manager is built,stop()throws before that, the workers of the owned leases start while the leases are loaded, outside the returnedMono). The consumer stops while the start is before the partition manager and the start then completes (the processor must be stopped); while the leases are loaded (stopped at once); while the next attempt builds a processor (it must not be started); a start fails after it started workers (stopped before the next processor starts); a restart after the processor started; a restart while the first start is in flight (the new processor receives the change, the old one is stopped when its start completes). Without the follow-up commit, 4 of the 6 fail (the lease-phase stop and the restart after a completed start also pass there; they guard that this still works).CosmosDbConsumerFirstStartFailsTestandCosmosDbConsumerStopWhileStartingTesteach passed 20 runs in a row in one JVM.How it was found: a review of consumers whose first connect or subscribe runs once and asynchronously, then the JUnit test.
What was not run: a test against Cosmos DB or its emulator, the ITs, the full root build.
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 and tested camel-azure-cosmosdb. No option changed. 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). Each commit carries a
Co-Authored-Bytrailer.Claude Code on behalf of allthingssecurity
🤖 Generated with Claude Code