[jira] [Work started] (CAMEL-24781) camel-kafka - support exactly-once (read-process-write) by sending consumer offsets inside the producer transaction

["Andrea Cosentino (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-24781?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Work on CAMEL-24781 started by Andrea Cosentino.
------------------------------------------------
> camel-kafka - support exactly-once (read-process-write) by sending consumer offsets inside the producer transaction
> -------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-24781
>                 URL: https://issues.apache.org/jira/browse/CAMEL-24781
>             Project: Camel
>          Issue Type: New Feature
>          Components: camel-kafka
>            Reporter: Andrea Cosentino
>            Assignee: Andrea Cosentino
>            Priority: Major
>
> h3. Problem
> Setting {{transacted=true}} on a Kafka producer today only wraps the producer {{send()}} calls in a Kafka transaction ({{initTransactions}} / {{beginTransaction}} / {{commitTransaction}} / {{abortTransaction}} in {{KafkaProducer}} and {{KafkaTransactionSynchronization}}). For a consume -> transform -> produce route ({{from("kafka:in")...to("kafka:out")}}) this is NOT exactly-once: the consumed offsets are committed by the separate consumer commit path (auto-commit or a {{CommitManager}}), *outside* the producer transaction. On failure this yields duplicates or offset/message divergence, even though users reasonably expect {{transacted=true}} to give read-process-write EOS.
> {{Producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata)}} is never called (it appears only in a Javadoc comment in {{KafkaConfiguration}}).
> h3. Proposed design (for discussion)
> Add an opt-in exactly-once mode for the consume->produce topology:
> * The consumer must stop committing offsets itself (disable auto-commit and the offset {{CommitManager}}) when EOS is enabled.
> * The consumer exposes, per exchange, its {{ConsumerGroupMetadata}} ({{consumer.groupMetadata()}} - already fetched in {{DefaultMetricsCollector}}) and the {{TopicPartition -> OffsetAndMetadata}} of the record(s) being processed (topic/partition/offset are already placed on the exchange).
> * Before {{commitTransaction()}}, the producer calls {{producer.sendOffsetsToTransaction(offsets, groupMetadata)}} so the offsets are committed atomically within the producer transaction.
> * A new option (e.g. {{exactlyOnce}} / or documenting the required combination) gates the behaviour; requires {{isolation.level=read_committed}} on downstream consumers.
> h3. Scope / risks
> * Crosses the consumer<->producer boundary in a route; needs a clean way to pass group metadata + offsets from the consumer endpoint to the producer endpoint (exchange properties, or a dedicated EOS coordinator).
> * Trust-sensitive transactional path; needs thorough tests (failure injection, rebalance) and an upgrade-guide entry.
> * Interaction with {{breakOnFirstError}}, batching, and manual commit must be defined.
> This is a design-first feature and should not be rushed; filing it to capture the gap and the proposed approach. Related transactional-robustness fix: CAMEL-24780.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)