gnodet-bot commented on code in PR #28637:
URL: https://github.com/apache/camel/pull/28637#discussion_r4407472876
##########
components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConsumer.java:
##########
@@ -46,27 +64,79 @@ public CosmosDbConsumer(final CosmosDbEndpoint endpoint,
final Processor process
@Override
protected void doInit() throws Exception {
super.doInit();
+ this.changeFeedProcessor = createChangeFeedProcessor();
+ }
+
+ // Visible for testing.
+ ChangeFeedProcessor createChangeFeedProcessor() {
this.clientWrapper = new
CosmosAsyncClientWrapper(getEndpoint().getCosmosAsyncClient());
- this.changeFeedProcessor =
getContainerOperations().captureEventsWithChangeFeed(
+ return getContainerOperations().captureEventsWithChangeFeed(
getLeaseContainerOperations().getContainer(), getHostName(),
this::onEventListener,
getConfiguration().getChangeFeedProcessorOptions());
}
@Override
protected void doStart() throws Exception {
super.doStart();
-
- // start our changeFeedProcessor
- changeFeedProcessor.start()
+ failedStarts = 0;
+
+ // start our changeFeedProcessor. 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: try
+ // again with a backoff until it starts or the consumer is stopped, or
the consumer stays started without ever
+ // receiving a change
+ startSubscription = Mono.defer(this::startChangeFeedProcessor)
+ // building a change feed processor blocks (block(), and with
the create*IfNotExists options it creates
+ // the containers), and the start fails on a thread of the
Cosmos DB client or of a Reactor scheduler,
+ // where blocking is not allowed: each attempt runs on a
thread that may block. A failure to build the
+ // processor is retried as a failure to start it
+ .subscribeOn(Schedulers.boundedElastic())
+ .retryWhen(Retry.backoff(Long.MAX_VALUE,
START_RETRY_MIN_BACKOFF)
+ .maxBackoff(START_RETRY_MAX_BACKOFF)
+ // one line per failed attempt, as it is tried again
for as long as it fails: the stack trace
+ // at DEBUG
+ .doBeforeRetry(signal -> {
+ String uri =
URISupport.sanitizeUri(getEndpoint().getEndpointUri());
+ LOG.warn("Cannot start the change feed processor
of {} (attempt {}), trying again: {}", uri,
+ signal.totalRetries() + 1,
signal.failure().toString());
+ LOG.debug("Cannot start the change feed processor
of {}", uri, signal.failure());
+ // try again with a new change feed processor: one
whose start failed after it built its
+ // partition manager cannot start again
(azure-cosmos marks its partition load balancer as
+ // started before the leases are initialized, and
every later start() throws "Partition
+ // load balancer already started"). The next
attempt builds it, on a thread that may block
+ changeFeedProcessor = null;
+ failedStarts = signal.totalRetries() + 1;
+ }))
+ .doOnSuccess(aVoid -> {
+ if (failedStarts > 0) {
+ LOG.info("Started the change feed processor of {}
after {} failed attempt(s)",
+
URISupport.sanitizeUri(getEndpoint().getEndpointUri()), failedStarts);
+ }
+ })
.subscribe(aVoid -> {
}, this::onErrorListener);
}
+ private Mono<Void> startChangeFeedProcessor() {
+ ChangeFeedProcessor processor = changeFeedProcessor;
+ if (processor == null) {
+ processor = createChangeFeedProcessor();
+ changeFeedProcessor = processor;
+ }
+ return processor.start();
+ }
+
@Override
protected void doStop() throws Exception {
- if (changeFeedProcessor != null) {
+ if (startSubscription != null) {
+ // no further attempt to start
+ startSubscription.dispose();
+ startSubscription = null;
+ }
+ // stop() fails on a change feed processor that has not started, for
example one that could not start yet
+ final ChangeFeedProcessor processor = changeFeedProcessor;
+ if (processor != null && processor.isStarted()) {
Review Comment:
⚠️ **Processor leak window: `isStarted()` race with async `start()`**
`doStop()` disposes `startSubscription` (cancelling future retry delays),
then reads `changeFeedProcessor` and stops it only if `isStarted()` is `true`.
However, `processor.start()` returns a `Mono<Void>` — it is asynchronous. If a
retry attempt is currently executing on `boundedElastic` and `start()` has been
called but not yet completed, `isStarted()` returns `false` at the time
`doStop()` checks it. The processor finishes starting *after* `doStop()`
returns, is never stopped, and leaks.
The scenario:
1. `doBeforeRetry` sets `changeFeedProcessor = null` (line 106)
2. A new `P_new` is created and assigned to `changeFeedProcessor`
3. `P_new.start()` is called on the boundedElastic thread
4. `doStop()` runs concurrently: `dispose()` cancels the retry delay, but
cannot interrupt the already-running `start()` call — `Disposable.dispose()` is
best-effort on threads
5. `doStop()` reads `changeFeedProcessor = P_new`, but `P_new.isStarted() =
false` (start in-flight) → skips `stop()`
6. `start()` completes → `P_new` is now running, never stopped
Fix — stop conservatively, catching the exception that the SDK throws when
the processor was never started:
```suggestion
final ChangeFeedProcessor processor = changeFeedProcessor;
if (processor != null) {
try {
// we wait until it stops; stop() throws on a processor that
could not start yet
processor.stop().block();
} catch (Exception e) {
LOG.debug("Ignoring stop() failure (processor did not
start): {}", e.toString());
}
}
```
Alternatively, track a `volatile boolean stopping` flag set in `doStop()`
before `dispose()`, checked in `startChangeFeedProcessor()` after creating the
new processor — if stopping, call `stop()` immediately and return an empty
`Mono`.
⚠️ **Pre-fix test validation:** Dynamic validation was skipped (no worktree
for this PR; tests require Azure SDK mocks). Static trace confirms all four
tests exercise the retryWhen path: the mock's first `start()` returns
`Mono.error()`, triggering `doBeforeRetry` which sets `changeFeedProcessor =
null`, and the subsequent attempt calls `createChangeFeedProcessor()` to build
a replacement. The `CountDownLatch` assertions would hang (and thus fail) in
the original code since `doStart()` only subscribed once. ✅
--
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]
