[
https://issues.apache.org/jira/browse/CAMEL-25861?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25861:
-----------------------------
Description:
{{CosmosDbConsumer.doStart}} (main 20cdb309f7d0, lines 56-63) subscribes once
to the start of the change feed processor:
{code:java}
changeFeedProcessor.start()
.subscribe(aVoid -> {
}, this::onErrorListener);
{code}
The start reads the lease container and the feed container (the SDK initializes
the collection properties and the lease store before it starts the partition
manager). So it fails when Cosmos DB cannot be reached when the route starts
(network or DNS not ready at boot, a short outage), when the lease container
does not exist yet (created by another deployment,
{{createLeaseContainerIfNotExists=false}} is the default) or when the role
assignment of a managed identity is not applied yet. {{onErrorListener}} hands
the failure to the exception handler, logged as "Error processing exchange",
and nothing starts the processor again: the route stays started, the consumer
never receives a change, and there is no health check that would show it. Only
a restart of the route helps. A second effect: the SDK's
{{ChangeFeedProcessor.stop()}} throws {{IllegalStateException: The
ChangeFeedProcessor instance has not fully started}} on a processor that has
not started, so stopping such a route fails in {{doStop}} (read in the SDK, not
run).
CAMEL-25857 (camel-kubernetes) is the same pattern: the first watch submitted
once, its failure never retried.
*Fix*: start the processor with {{Mono.defer(() ->
changeFeedProcessor.start()).retryWhen(Retry.backoff(Long.MAX_VALUE, 1
s).maxBackoff(30 s))}}, logging a WARN with the cause for each failed attempt,
and use a new change feed processor for each attempt after a failure;
{{doStop}} disposes the pending attempts and calls {{stop()}} only on a
processor that {{isStarted()}}. A new processor is needed because the SDK's
{{start()}} can be called again only while the processor has no partition
manager: 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. 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()}} of that processor
throws {{IllegalStateException: Partition load balancer already started}}.
*Tests*: new {{CosmosDbConsumerFirstStartFailsTest}}: a change feed processor
whose first {{start()}} fails ({{Resource Not Found}}) and whose second start
delivers one change. On main it fails after 20 s: {{the change feed processor
was started 1 time(s), the consumer is started but receives nothing}}. With the
fix the change is received about 1 s after the WARN. A second test,
{{consumerStartsANewChangeFeedProcessorWhenTheFirstOneFailedWhileInitializingTheLeases}},
gives a first processor that fails its start and then throws {{Partition load
balancer already started}} as the SDK's does; retrying that same processor
fails after 20 s ({{the failed change feed processor was started 5 time(s), no
new one was started}}), with the fix a second processor starts and the change
is received. camel-azure-cosmosdb unit tests with the fix: 23 run, 0 failures.
*How it was found*: a review of consumers whose first connect or subscribe runs
once and asynchronously (the CAMEL-25857 pattern), then the JUnit test. Not run
against Cosmos DB or its emulator.
*Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
(the same single subscription in {{doStart}}; read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
was:
{{CosmosDbConsumer.doStart}} (main 20cdb309f7d0, lines 56-63) subscribes once
to the start of the change feed processor:
{code:java}
changeFeedProcessor.start()
.subscribe(aVoid -> {
}, this::onErrorListener);
{code}
The start reads the lease container and the feed container (the SDK initializes
the collection properties and the lease store before it starts the partition
manager). So it fails when Cosmos DB cannot be reached when the route starts
(network or DNS not ready at boot, a short outage), when the lease container
does not exist yet (created by another deployment,
{{createLeaseContainerIfNotExists=false}} is the default) or when the role
assignment of a managed identity is not applied yet. {{onErrorListener}} hands
the failure to the exception handler, logged as "Error processing exchange",
and nothing starts the processor again: the route stays started, the consumer
never receives a change, and there is no health check that would show it. Only
a restart of the route helps. A second effect: the SDK's
{{ChangeFeedProcessor.stop()}} throws {{IllegalStateException: The
ChangeFeedProcessor instance has not fully started}} on a processor that has
not started, so stopping such a route fails in {{doStop}} (read in the SDK, not
run).
CAMEL-25857 (camel-kubernetes) is the same pattern: the first watch submitted
once, its failure never retried.
*Fix*: start the processor with
{{Mono.defer(changeFeedProcessor::start).retryWhen(Retry.backoff(Long.MAX_VALUE,
1 s).maxBackoff(30 s))}}, logging a WARN with the cause for each failed
attempt; {{doStop}} disposes the pending attempts and calls {{stop()}} only on
a processor that {{isStarted()}}. The SDK's {{start()}} can be called again
after a failure (it builds the partition manager again while it has none).
*Tests*: new {{CosmosDbConsumerFirstStartFailsTest}}: a change feed processor
whose first {{start()}} fails ({{Resource Not Found}}) and whose second start
delivers one change. On main it fails after 20 s: {{the change feed processor
was started 1 time(s), the consumer is started but receives nothing}}. With the
fix the change is received about 1 s after the WARN. camel-azure-cosmosdb unit
tests with the fix: 22 run, 0 failures.
*How it was found*: a review of consumers whose first connect or subscribe runs
once and asynchronously (the CAMEL-25857 pattern), then the JUnit test. Not run
against Cosmos DB or its emulator.
*Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
(the same single subscription in {{doStart}}; read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
> camel-azure-cosmosdb - when the change feed processor cannot start (Cosmos DB
> not reachable yet, lease container or role assignment not there yet) the
> consumer never starts it again and receives no change while the route is
> started
> ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25861
> URL: https://issues.apache.org/jira/browse/CAMEL-25861
> Project: Camel
> Issue Type: Bug
> Components: camel-azure
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> {{CosmosDbConsumer.doStart}} (main 20cdb309f7d0, lines 56-63) subscribes once
> to the start of the change feed processor:
> {code:java}
> changeFeedProcessor.start()
> .subscribe(aVoid -> {
> }, this::onErrorListener);
> {code}
> The start reads the lease container and the feed container (the SDK
> initializes the collection properties and the lease store before it starts
> the partition manager). So it fails when Cosmos DB cannot be reached when the
> route starts (network or DNS not ready at boot, a short outage), when the
> lease container does not exist yet (created by another deployment,
> {{createLeaseContainerIfNotExists=false}} is the default) or when the role
> assignment of a managed identity is not applied yet. {{onErrorListener}}
> hands the failure to the exception handler, logged as "Error processing
> exchange", and nothing starts the processor again: the route stays started,
> the consumer never receives a change, and there is no health check that would
> show it. Only a restart of the route helps. A second effect: the SDK's
> {{ChangeFeedProcessor.stop()}} throws {{IllegalStateException: The
> ChangeFeedProcessor instance has not fully started}} on a processor that has
> not started, so stopping such a route fails in {{doStop}} (read in the SDK,
> not run).
> CAMEL-25857 (camel-kubernetes) is the same pattern: the first watch submitted
> once, its failure never retried.
> *Fix*: start the processor with {{Mono.defer(() ->
> changeFeedProcessor.start()).retryWhen(Retry.backoff(Long.MAX_VALUE, 1
> s).maxBackoff(30 s))}}, logging a WARN with the cause for each failed
> attempt, and use a new change feed processor for each attempt after a
> failure; {{doStop}} disposes the pending attempts and calls {{stop()}} only
> on a processor that {{isStarted()}}. A new processor is needed because the
> SDK's {{start()}} can be called again only while the processor has no
> partition manager: 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.
> 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()}} of that
> processor throws {{IllegalStateException: Partition load balancer already
> started}}.
> *Tests*: new {{CosmosDbConsumerFirstStartFailsTest}}: a change feed processor
> whose first {{start()}} fails ({{Resource Not Found}}) and whose second start
> delivers one change. On main it fails after 20 s: {{the change feed processor
> was started 1 time(s), the consumer is started but receives nothing}}. With
> the fix the change is received about 1 s after the WARN. A second test,
> {{consumerStartsANewChangeFeedProcessorWhenTheFirstOneFailedWhileInitializingTheLeases}},
> gives a first processor that fails its start and then throws {{Partition
> load balancer already started}} as the SDK's does; retrying that same
> processor fails after 20 s ({{the failed change feed processor was started 5
> time(s), no new one was started}}), with the fix a second processor starts
> and the change is received. camel-azure-cosmosdb unit tests with the fix: 23
> run, 0 failures.
> *How it was found*: a review of consumers whose first connect or subscribe
> runs once and asynchronously (the CAMEL-25857 pattern), then the JUnit test.
> Not run against Cosmos DB or its emulator.
> *Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
> (the same single subscription in {{doStart}}; read, not run).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
