oscerd opened a new pull request, #25617:
URL: https://github.com/apache/camel/pull/25617
# CAMEL-24382: camel-azure-cosmosdb — change-feed consumer loses events on
failure
## Problem
`CosmosDbConsumer.onEventListener` dispatched the exchange
**fire-and-forget** and returned immediately:
```java
exchange.getExchangeExtension().addOnCompletion(onCompletion);
getAsyncProcessor().process(exchange, EmptyAsyncCallback.get());
```
The change feed is driven by the Azure SDK `ChangeFeedProcessor`, built with
a **synchronous** `handleChanges(Consumer<List<...>>)` handler
(`CosmosDbContainerOperations.captureEventsWithChangeFeed`). The SDK advances
(checkpoints) the lease **as soon as that handler returns** — which happened as
soon as the exchange had been *queued*, before the route actually processed it.
`ConsumerOnCompletion` only implements `onFailure` (it logs). So on any route
failure the lease had already advanced and the batch was **never redelivered**:
the consumer was effectively at-most-once and silently lost data on downstream
failures (or on shutdown mid-batch).
## Fix
Process the batch **synchronously** inside the handler and rethrow on
failure, so the lease is only checkpointed after successful processing:
```java
try {
getProcessor().process(exchange);
} catch (Exception e) {
exchange.setException(e);
}
if (exchange.getException() != null) {
throw new RuntimeCamelException("Error processing Azure CosmosDB change
feed batch", ...);
}
```
When the exchange fails, the handler throws, the `ChangeFeedProcessor` does
**not** advance the lease, and the batch is redelivered on the next feed poll
(**at-least-once**). Successful batches return normally and are checkpointed as
before. Processing now blocks the per-lease feed thread until the exchange
completes, which is the required behaviour for at-least-once delivery
(partitions still process in parallel).
## Test
The batch-processing logic is extracted into a package-private
`processBatch(Exchange)` seam and `CosmosDbConsumerTest` covers both paths
without needing a Cosmos emulator: a failing route causes `processBatch` to
throw `RuntimeCamelException` (so the lease is not advanced), and a successful
route returns normally. The consumer's happy path also remains covered by the
emulator-driven `CosmosDbConsumerIT`. Full reactor build is green.
_Claude Code on behalf of oscerd_
--
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]
