[
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)
