Andrea Cosentino created CAMEL-24781:
----------------------------------------
Summary: 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
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)
