oscerd opened a new pull request, #26557: URL: https://github.com/apache/camel/pull/26557
> **Draft / first cut for design review.** This proposes read-process-write exactly-once for Kafka-to-Kafka routes. The code path is functional and unit-tested, but the design has open questions (below) and some mechanical follow-ups (DSL regen, IT). Feedback on the approach is very welcome before this is finished. ## Problem Setting `transacted=true` on a Kafka producer today only wraps the producer `send()`s in a Kafka transaction. For a consume → transform → produce route the **consumer offsets are committed on a separate path** (auto-commit or a `CommitManager`), *outside* the producer transaction — so it is **not** exactly-once: a failure can duplicate or diverge messages vs offsets, even though `transacted=true` implies EOS. `Producer.sendOffsetsToTransaction(...)` was never called. ## Approach Add an opt-in producer option **`exactlyOnce`**. When enabled together with a transactional producer, `KafkaTransactionSynchronization` calls `Producer.sendOffsetsToTransaction(offsets, groupMetadata)` **before** `commitTransaction()`, so the consumed record's offset and the produced records are committed atomically. The offsets + group metadata are taken from the `KafkaManualCommit` object the source Kafka consumer already places on the exchange (`CamelKafkaManualCommit` header). `DefaultKafkaManualCommit` gains `getConsumerGroupMetadata()`; the producer reads it **on the consumer poll thread** (the Kafka consumer is not thread-safe) while building the synchronization, and the offset committed is `record offset + 1`. ## Required configuration - Source consumer: `allowManualCommit=true` **and** `autoCommitEnable=false` (so it does not commit offsets itself and the manual-commit object is on the exchange). - Producer: `transacted=true` (or `transactionalId=...`) **and** `exactlyOnce=true`. - Downstream consumers: `isolation.level=read_committed`. ## What's included - `KafkaConfiguration.exactlyOnce` option (+ generated catalog metadata). - `KafkaTransactionSynchronization` sends offsets into the transaction before commit. - `KafkaProducer.createTransactionSynchronization` resolves the offsets/group metadata from the manual-commit object (poll thread). - `DefaultKafkaManualCommit.getConsumerGroupMetadata()`. - Unit test `KafkaTransactionSynchronizationTest` (offsets sent before commit; not sent when not EOS). - Also begins the broker transaction before marking the unit of work (aligns with CAMEL-24780). ## Open design questions 1. **Vehicle**: reuse the consumer's `KafkaManualCommit` (as here) vs a dedicated EOS token on the exchange? 2. **Consumer commit disabling**: should EOS mode enforce `autoCommitEnable=false` automatically rather than requiring the user to set it? 3. **Threading**: group metadata must be read on the poll thread, so the produce step must run synchronously on the consumer thread (no async handoff before the producer). Should this be validated/enforced? 4. **Scope**: this first cut covers the streaming single-record path; batching and multi-record-per-transaction accumulation are not handled yet. ## Follow-ups before merge - Endpoint-DSL / component-DSL builder regeneration for `exactlyOnce()` (mechanical; not run locally — CI will flag). - End-to-end integration test (`from(kafka).to(kafka)` transacted, verifying atomic offset+message commit and behaviour on failure/rebalance). - Component docs / example. JIRA: https://issues.apache.org/jira/browse/CAMEL-24781 _Claude Code on behalf of @oscerd_ 🤖 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]
