Andrea Cosentino created CAMEL-25221:
----------------------------------------
Summary: camel-couchbase: consumerProcessedStrategy=delete removes
the document before the route runs, so an undelivered row is lost
Key: CAMEL-25221
URL: https://issues.apache.org/jira/browse/CAMEL-25221
Project: Camel
Issue Type: Bug
Components: camel-couchbase
Reporter: Andrea Cosentino
Raised by davsclaus reviewing [PR
#27123|https://github.com/apache/camel/pull/27123]. This predates that change
and is not fixed by it.
With {{consumerProcessedStrategy=delete}}, {{CouchbaseConsumer}} removes each
document *while it is building the exchanges*, in {{pollWithSqlQuery}} and
{{pollWithView}}:
{code:java}
for (JsonObject row : result.rowsAsObject()) {
...
Exchange exchange = createExchange(true);
...
if ("delete".equalsIgnoreCase(consumerProcessedStrategy)) {
CouchbaseCollectionOperation.removeDocument(collection, id,
endpoint.getWriteQueryTimeout(),
endpoint.getConsumerRetryPause());
}
...
exchanges.add(exchange);
}
return processBatch(exchanges);
{code}
The delete therefore happens *before* {{processBatch}} hands anything to the
route. Two consequences:
h2. A row that is never delivered is still deleted
{{processBatch}} stops at {{maxMessagesPerPoll}} and at {{!isBatchAllowed()}}.
Every row past that point has already been removed from Couchbase but was never
given to the route, and the next poll cannot see it again. The documents are
simply gone.
{{isBatchAllowed()}} returns false once the consumer is stopping, so this is
reachable in a default configuration: stopping a route mid-batch loses the
remainder. ({{maxMessagesPerPoll}} is a second route to it, though note it is
not currently declared as an endpoint option on this component, so it can only
be set programmatically.)
h2. A failed exchange is also a lost document
Even for rows that *are* delivered, the document is gone by the time the route
runs, so a route failure loses it. PR #27123 makes that failure visible (the
consumer now reports it through its exception handler instead of discarding the
outcome), but visibility is not recovery.
h2. Options
* Delete *after* the exchange has been processed successfully - which is what
the option name implies, and matches how the strategy reads in the docs.
* Or cap the query itself at {{maxMessagesPerPoll}} so no row is fetched, and
therefore deleted, unless it will be handed to the route. This does not address
the failed-route case.
The first is the real fix; it changes when the delete happens, so it wants an
upgrade-guide note.
For reference, the comparable change in {{camel-mongodb}} was CAMEL-25024,
where the position is advanced only after a successful exchange.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
