[
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)
