allthingssecurity commented on code in PR #28637:
URL: https://github.com/apache/camel/pull/28637#discussion_r4408160706
##########
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:
Thanks, this was real. Fixed in d1af39ff2c0b (one new commit on top).
Neither of the two suggestions was enough on its own, because of what
azure-cosmos 4.75 does. I checked this with javap on the
`pkversion.IncrementalChangeFeedProcessorImpl` that `handleChanges` builds.
What the SDK does:
- Until the processor has built its partition manager, `isStarted()` is
false and `stop()` throws `IllegalStateException("The ChangeFeedProcessor
instance has not fully started")` synchronously. Stopping unconditionally and
catching only skips the stop. If the start then goes on, the processor runs and
nobody stops it.
- After that, `isStarted()` is true while the leases are still loading, and
`stop()` works.
- While the leases load, it starts a worker for each lease this host owns,
outside the `Mono` that `start()` returns. If the start then fails, for example
with a 429 on a later lease, those workers keep running. The previous head
dropped such a processor and built a new one, so the old workers ran next to
the new ones.
- A boolean `stopping` flag is false again after a restart, so an old
attempt would take over the new processor. It has to be a generation.
So now the processor, the attempts and `doStop` hand over under a lock, with
a generation that `doStart` and `doStop` increment:
- `doStop` no longer cancels a start that is in flight. It stops the
processor if it can already.
- The attempt always sees the end of the start, which is signalled under the
lock. If the consumer stopped meanwhile, the attempt stops the processor then.
- A start that fails is stopped too, if it had started.
- The attempt calls `stop().subscribe(...)`; only `doStop` blocks, on the
Camel thread, as before. Nothing calls `block()` on a Reactor thread.
| Stop comes | What happens | Test |
|---|---|---|
| before the retry builds (backoff pending) | `dispose()` cancels the delay.
An attempt that runs anyway sees the new generation and builds nothing. | the
generation check |
| while it builds, or between the build and `start()` | `dispose()`
interrupts the build. After the build, the attempt sees the new generation and
does not call `start()`. An interrupted build is not reported as a failure. |
`consumerDoesNotStartAChangeFeedProcessorBuiltAfterTheConsumerStopped` |
| while `start()` is in flight, before the partition manager exists |
`doStop` cannot stop it, because `stop()` throws. When the start ends, the
attempt stops it. |
`consumerStopsAChangeFeedProcessorThatFinishesStartingAfterTheConsumerStopped` |
| while `start()` is in flight, leases loading (`isStarted()` true) |
`doStop` stops it at once. Workers scheduled after that get the cancelled
token. | `consumerStopsAChangeFeedProcessorThatIsLoadingTheLeasesAtOnce` |
| after the start completed | The success was signalled before `doStop`,
which calls `stop().block()` as before. | the existing tests |
| stop, then start again | A restart runs `doInit` again, which builds a new
processor (as before). A processor whose start was still in flight is stopped
by its attempt when its start ends. The old attempt cannot touch the new
generation. | `consumerStartsTheChangeFeedProcessorAgainWhenRestarted`,
`consumerStartsANewChangeFeedProcessorWhenRestartedWhileTheOldOneIsStarting` |
Also
`consumerStopsAChangeFeedProcessorWhoseStartFailedWhileLoadingTheLeases` covers
the failed start that left workers running.
The new tests use a processor that behaves as the 4.75 one: `isStarted()`
once the partition manager is built, `stop()` throws before that, and the
workers start outside the returned `Mono`. On the previous head, 4 of the 6
fail. The other 2 also pass there: the lease-phase stop and the plain restart.
They guard that this still works.
One window is left. If the consumer stops before the partition manager
exists and the start then goes on, the workers of leases this host already owns
can run until the start ends. The attempt then stops them. Nothing can stop
them earlier, since `stop()` throws until then.
_Claude Code on behalf of allthingssecurity_
##########
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:
I kept the handler and added a comment instead (in d1af39ff2c0b). It is not
quite unreachable. The retries never run out, but the chain still ends with an
error if the retry itself fails, for example when a scheduler rejects the delay
before the next attempt. The consumer then receives nothing, so the exception
handler is the right place for it.
Making it reachable with a bounded number of retries would bring back the
bug this PR fixes: a started consumer that gave up and never receives a change.
_Claude Code on behalf of allthingssecurity_
--
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]
