gnodet-bot commented on code in PR #28637:
URL: https://github.com/apache/camel/pull/28637#discussion_r4407479255
##########
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);
Review Comment:
💡 **`onErrorListener` is unreachable with `Long.MAX_VALUE` retries**
`Retry.backoff(Long.MAX_VALUE, ...)` never exhausts its retry count, so no
terminal error is ever propagated downstream to the `subscribe()` error handler
— `this::onErrorListener` can never be called. The only way the chain
terminates is through `dispose()` (from `doStop()`), which completes without an
error signal.
Consider adding a comment to avoid confusion:
```suggestion
.subscribe(aVoid -> {
}, this::onErrorListener); // defensive: unreachable with
Long.MAX_VALUE retries
```
--
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]
