shashank created CAMEL-26822:
--------------------------------
Summary: camel-kafka - the Kafka idempotent and key-value
repositories force acks=1, so a leader failover can drop acknowledged records
Key: CAMEL-26822
URL: https://issues.apache.org/jira/browse/CAMEL-26822
Project: Camel
Issue Type: Improvement
Components: camel-kafka
Reporter: shashank
Assignee: shashank
{{KafkaChangelog.start}} sets
{{producerConfig.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1")}}
({{KafkaChangelog.java:162}}). {{KafkaIdempotentRepository}} and
{{KafkaKeyValueRepository}} both use {{KafkaChangelog}}, so every repository
whose producer config does not set {{acks}} runs with {{acks=1}}. This includes
{{new KafkaIdempotentRepository(topic, bootstrapServers)}}.
Since Kafka 3.0 (KIP-679) the client default is {{acks=all}} with
{{enable.idempotence=true}}. The line overrides that, and {{acks=1}} also turns
the default idempotence off (the client logs "Idempotence will be disabled
because acks is set to 1"). A producer config that sets only
{{enable.idempotence=true}} fails at start with a {{ConfigException}}.
With {{acks=1}}, {{KafkaChangelog.send}} counts the leader's acknowledgement as
success, and the repository has already changed its local cache. If the leader
fails over before a follower has the record, the record is truncated. No
instance ever read it, so the writer's cache and the topic disagree for good.
Example, replication factor 3, two instances:
# Instance A adds {{k1}}. The add is replicated and both instances apply it.
# A's processing fails and {{remove(k1)}} drops {{k1}} from A's cache. Only the
leader has acknowledged the remove record.
# The leader stops (a crash, or a controlled shutdown in a rolling restart). A
follower that was in sync becomes leader without the remove record.
# The redelivery of {{k1}} reaches B, which still holds {{k1}}, and is skipped
as a duplicate. A also skips it after a restart.
A lost add lets a duplicate through, and a lost {{put}} of the key-value
repository makes readers go back to the old value.
Found with Specula (a TLA+ model checked against traces of the real code). The
failover itself was not reproduced with a test: it needs a cluster of several
brokers and a leader change.
Why it matters: the default settings can hit it on any cluster with
replication, on a normal rolling restart. The docs set no requirement on the
replication factor, and the {{KafkaKeyValueRepository}} javadoc
({{KafkaKeyValueRepository.java:52}}) calls it "durable" storage.
Workaround: set {{acks=all}} in the repository's {{producerConfig}}. The line
uses {{putIfAbsent}}, so a user value wins.
Background: the line dates from the first implementation (CAMEL-10927, 2017),
when the client default was already {{acks=1}}. Its comment is about
synchronous sends without batching. CAMEL-18841 changed the endpoint default
{{requestRequiredAcks}} to {{all}}, but the repositories were not changed.
Suggested change: remove the {{acks}} line so the client default applies, and
add a note to the upgrade guide. Would you prefer that, or setting {{acks=all}}
explicitly? Each send then waits for the in-sync replicas, which adds some
latency to every add and remove.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
