allthingssecurity opened a new pull request, #29587: URL: https://github.com/apache/camel/pull/29587
# Description [CAMEL-26822](https://issues.apache.org/jira/browse/CAMEL-26822) `KafkaChangelog`, which `KafkaIdempotentRepository` and `KafkaKeyValueRepository` write their changes with, set `producerConfig.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1")` on its producer. So every repository whose producer config does not set `acks` (including `new KafkaIdempotentRepository(topic, bootstrapServers)`) ran with `acks=1`: - it overrode the default of the Kafka client, which since Kafka 3.0 (KIP-679) is `acks=all` with `enable.idempotence=true`, and turned that idempotence off; - a producer config that sets only `enable.idempotence=true` failed at start with a `ConfigException` ("Must set acks to all in order to use the idempotent producer"); - with `acks=1`, `KafkaChangelog.send` counts the leader's acknowledgement as success, and the repository has already changed its local cache. When the leader fails over before a follower has the record, the record is truncated, and the writer's cache and the topic disagree for good (a lost add lets a duplicate through on the other instances; a lost remove makes them skip a redelivery; a lost `put` makes readers go back to the old value). The line dates from the first implementation (CAMEL-10927, 2017), when the client default was `acks=1`; its comment is about synchronous sends without batching, which stays (`batch.size=0`). CAMEL-18841 changed the endpoint default `requestRequiredAcks` to `all`, but not the repositories. Change: the line is removed, so the client default applies (`acks=all` with idempotence); a producer config that sets `acks` still wins. The JIRA asked whether to remove the line or set `acks=all` explicitly: removing it follows the client (and the endpoint producers, whose default is `acks=all` with idempotence already), and also gives the idempotence, which an explicit `acks=all` would leave to the client too. The preparation of the producer properties moves into a package-private method so it can be tested without a broker. Each change of a repository now waits until the in-sync replicas have it (the sends are synchronous). With fewer in-sync replicas than `min.insync.replicas`, a change now waits up to `delivery.timeout.ms` (2 minutes by default) and then fails, where it succeeded with `acks=1`; a broker before Kafka 2.8 that uses ACLs needs the `IdempotentWrite` permission for the producer. The 4.24 upgrade guide says so, and how to keep the previous behaviour: `acks=1` in the `producerConfig`, together with `bootstrap.servers`, as a `producerConfig` replaces the `bootstrapServers` of the repository for the producer. Main only (a behaviour change). Tests, new `KafkaChangelogProducerConfigTest` (no broker: the Kafka client resolves `ProducerConfig` from the properties the changelog builds): - `producerUsesTheAcksOfTheKafkaClient`: acks resolves to all (`-1`) and idempotence is on; - `producerConfigThatEnablesIdempotenceIsValid`: a producer config with only `enable.idempotence=true` is accepted; - `acksOfTheProducerConfigWins`: a control, `acks=1` from the producer config is kept (passes before the change too). With the old line (and the same method extracted), the first two fail, with the `ConfigException` above for the second. A test of the failover itself needs a cluster of several brokers and a controlled leader change, which the unit tests do not have; the JIRA describes that scenario. With the change the 3 pass, and the camel-kafka unit tests pass (321 run, 0 failures, 1 skipped). # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-26822](https://issues.apache.org/jira/browse/CAMEL-26822). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built the core modules and camel-kafka with the formatter and import-sort plugins and ran the camel-kafka unit tests; no generated file changes. I did not run the full root build. The ITs, which need a Kafka container, were not run locally; CI runs them.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
