[jira] [Resolved] (CAMEL-24953) Idempotent Consumer with skipDuplicate=false: a duplicate that fails removes the message id that the original exchange added

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-24953?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen resolved CAMEL-24953.
---------------------------------
    Fix Version/s: 4.23.0
       Resolution: Fixed

The fix is merged on main, so it is in Camel 4.23.0:
* 64e17e003b0c CAMEL-24953: camel-core - Idempotent Consumer: do not let a routed duplicate remove or confirm the key
* c0291dc08f03 CAMEL-24953: camel-core - Make the new test class package-private
* f80b68052fce CAMEL-24953: camel-core - Use KeyValueIdempotentRepository in the new test and always release the blocked exchange

Resolving, as the ticket was not updated when the PR was merged.

_Claude Code on behalf of Claus Ibsen_

> Idempotent Consumer with skipDuplicate=false: a duplicate that fails removes the message id that the original exchange added
> ----------------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-24953
>                 URL: https://issues.apache.org/jira/browse/CAMEL-24953
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-core
>            Reporter: shashank
>            Priority: Minor
>             Fix For: 4.23.0
>
>
> With {{skipDuplicate(false)}}, a duplicate is not dropped. It is flagged with {{CamelDuplicateMessage=true}} and continues through the route, typically into a {{filter}} or {{choice}} that sends duplicates somewhere else. {{IdempotentConsumer.process}} registers the same {{IdempotentOnCompletion}} for a duplicate as for a new message, and the completion does not know that the exchange is a duplicate. If the duplicate fails, {{onFailedMessage}} calls {{idempotentRepository.remove(messageId)}} (with the default {{removeOnFailure=true}}). That deletes the entry the *original* exchange added, whether the original has already completed successfully or is still being processed.
> Result:
> * The next copy of the message is processed again, although the original was processed successfully.
> * In eager mode (the default), that next copy can be processed at the same time as the original, which is the case eager mode is meant to prevent.
> Reproducer (the same result with {{MemoryIdempotentRepository}}, {{KeyValueIdempotentRepository}} and {{FileIdempotentRepository}}):
> {code:java}
> from("direct:in")
>     .idempotentConsumer(header("id")).idempotentRepository(repo).skipDuplicate(false)
>     .choice()
>         .when(exchangeProperty(Exchange.DUPLICATE_MESSAGE).isEqualTo(true))
>             .process(e -> { throw new IllegalStateException("duplicate audit endpoint down"); })
>         .otherwise()
>             .process(e -> processed.incrementAndGet())
>     .end();
> {code}
> Sequential sends with header {{id=1}}:
> # m1 is processed. {{repo.contains("1")}} is true.
> # m2 is flagged as a duplicate, and the duplicate branch throws. After this, {{repo.contains("1")}} is *false*.
> # m3 is *processed as a new message* (processed=2, expected 1).
> Concurrent variant: m1 is held inside the route by a latch, m2 is a duplicate that fails, and m3 is processed as new while m1 is still in flight. Two exchanges with the same id are processed concurrently in eager mode.
> The javadoc describes the removal as a rollback of the failing exchange's own add. The {{IdempotentRepository}} javadoc says "On failure, remove(String) rolls back so the message can be redelivered", and the EIP page says the id is removed "if the Exchange failed, otherwise it stays there". A duplicate did not add the id, so there is nothing for it to roll back. For the same reason, a successful duplicate should not {{confirm}} (or, in non-eager mode, {{add}}) the original's key. That is a no-op in the core repositories, but not necessarily in others.
> Proposed fix: in {{IdempotentConsumer.process}}, only register the {{IdempotentOnCompletion}} for a new message. A duplicate that is not skipped ({{skipDuplicate=false}}) still gets the {{CamelDuplicateMessage=true}} property, still increments the duplicate counter and still calls {{onDuplicateMessage}}, and is then routed on with the original callback and no idempotent completion. So a duplicate never calls {{add}}, {{confirm}} or {{remove}} on the key. New messages behave as before, for every combination of {{eager}}, {{completionEager}} and {{removeOnFailure}}. A PR with the fix follows, with regression test {{IdempotentConsumerFailedDuplicateTest}} (sequential case with eager, completionEager and non-eager, plus the concurrent eager case); all four tests fail without the change.
> Found with a TLA+ model of the Idempotent Consumer EIP, then reproduced against 4.23.0-SNAPSHOT. The model with this change keeps the invariants "an id that was processed successfully is not processed again", "a remove only rolls back the exchange's own add" and "eager mode never processes two exchanges with the same id concurrently".
> Related question, non-eager mode: with {{eager(false)}}, two exchanges with the same id can both pass {{contains()}}, which is documented. If one succeeds (and adds the id) and the other then fails, the failing one removes the id, and a later copy is processed again. The {{IdempotentRepository}} javadoc does say that {{remove}} "is still called on failure" in non-eager mode. But a non-eager exchange that fails never added the id, so the call can only remove an entry that another exchange committed. Should {{removeOnFailure}} apply in non-eager mode at all? The fix leaves this as it is (it is documented behaviour).



--
This message was sent by Atlassian Jira
(v8.20.10#820010)