allthingssecurity opened a new pull request, #28637: URL: https://github.com/apache/camel/pull/28637
# Description [CAMEL-25861](https://issues.apache.org/jira/browse/CAMEL-25861) `CosmosDbConsumer` subscribed once to `ChangeFeedProcessor.start()` in `doStart`. 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 (`createLeaseContainerIfNotExists` is `false` by 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's `stop()` throws on a processor that has not started. (With `createDatabaseIfNotExists`, `createContainerIfNotExists` or their lease variants, building the processor in `doInit` already 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: - `doStart` starts the processor with `Mono.defer(() -> changeFeedProcessor.start()).retryWhen(Retry.backoff(...))`: 1 s, then doubling up to 30 s, until it starts or the consumer is stopped. Each 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). The failed attempts are no longer passed to the exception handler of the consumer, which got the one failure before. - Each attempt after a failure uses a new change feed processor. A processor whose start failed after it built its partition manager cannot start again: in azure-cosmos 4.75.0 `PartitionManagerImpl.start()` calls `PartitionLoadBalancerImpl.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 later `start()` throws `IllegalStateException: 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. - The attempt builds the new processor, not the retry hook, and every attempt runs on Reactor's `boundedElastic` scheduler (`subscribeOn` before `retryWhen`, so each resubscription goes through it). Building a processor blocks: `CosmosDbContainerOperations.captureEventsWithChangeFeed` zips the feed and lease containers and calls `block()` (with the `create*IfNotExists` options, 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, where `block()` throws `IllegalStateException` (reactor-core 3.8.7 `BlockingSingleSubscriber.blockingGet` checks `Schedulers.isInNonBlockingThread()` first); 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. - The change feed processor is created in a package-private `createChangeFeedProcessor()`, which the test overrides (instead of setting the private field by reflection). - `doStop` cancels the pending attempts, and calls `stop()` only on a processor that `isStarted()`. No option changed, so no upgrade guide entry. Tests: - New `CosmosDbConsumerFirstStartFailsTest` (4 tests). `consumerStartsTheChangeFeedAgainWhenTheFirstStartFails`: a change feed processor whose first `start()` fails and whose second start delivers one change. Without the change it fails after 20 s with `the 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 throws `Partition load balancer already started` as 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 Reactor `parallel` thread (`Mono.delay(...).then(Mono.error(...))`), and the test's `createChangeFeedProcessor()` ends in `Mono.zip(...).block()` as the real one does. With a retry hook that builds the processor (the first version of this change) the hook runs on `parallel-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]`). - `consumerTriesAgainWhenANewChangeFeedProcessorCannotBeBuilt` (same class): after the first start fails, building the next processor throws once (as with `createLeaseContainerIfNotExists=true` while 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. - camel-azure-cosmosdb unit tests with the change: 29 run, 0 failures (22 ITs skipped, as without the credentials). 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 - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-25861](https://issues.apache.org/jira/browse/CAMEL-25861). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally 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 - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) 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-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
