[
https://issues.apache.org/jira/browse/CAMEL-26822?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26822 started by shashank.
----------------------------------------
> 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
> Priority: Major
>
> {{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)
