shashank created CAMEL-24953:
--------------------------------
Summary: 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
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)
