ramu11 commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4451392943
##########
components/camel-kafka/src/main/docs/kafka-component.adoc:
##########
@@ -552,10 +555,157 @@ At the end of exchange routing, the kafka producer would
commit the transaction
It would work with JTA `camel-jta` by using `transacted()` and if it involves
some resources (SQL or JMS), which supports XA, then they would work in tandem,
where they both will either commit or rollback at the end of the exchange
routing. In some cases, if the JTA transaction manager fails to commit (during
the 2PC processing), but kafka transaction has been committed before and there
is no chance to roll back the changes since the kafka transaction does not
support JTA/XA spec. There is still a risk with the data consistency.
-When you use the `transacted=true` parameter, the `transactionalId` is made
from the endpoint id and route id, but if you want finer control you can just
use the `transactionalId` parameter, however remeber to have a
`transactionalId` for each kafka producer endpoint. Note that the old style of
using `additionalProperties[transactional.id]` is still valid.
+When you use the `transacted=true` parameter, the `transactionalId` is made
from the endpoint ID and route ID.
+If you want finer control, set the `transactionalId` parameter; use a
different ID for each producer endpoint.
+The old style of using `additionalProperties[transactional.id]` is still
valid. For exactly-once read-process-write,
+the ID must be unique across application replicas and stable across restarts
of the same producer instance; the
+generated endpoint/route ID alone does not meet that requirement.
If both `transacted=true` and `transactionalId` are present, the latter takes
precedence.
+=== Kafka exactly-once (read-process-write)
+
+The Kafka component can atomically commit records produced by a route and the
source consumer offsets. This is an
+opt-in feature for Kafka-to-Kafka routes. Configure `exactlyOnce=true` on both
Kafka endpoints:
+
+* Source consumer: `exactlyOnce=true`, `allowManualCommit=true`,
`autoCommitEnable=false` and
+ `breakOnFirstError=true`. Automatic commits are disabled. A transactional
destination producer commits the source
+ offsets with its output records; when a successful route branch does not use
a Kafka producer (for example, a filter
+ drops the record), Camel commits the source offset directly on the polling
consumer thread.
+* Destination producer: `exactlyOnce=true` and an explicit `transactionalId`
+ (or `additionalProperties[transactional.id]`).
+ `transacted=true` alone is not sufficient in this mode. Treat the configured
`transactionalId` as a base ID: Camel
+ appends the owning route ID when it creates the producer
(`<base>-<routeId>`). This gives separate producer instances
+ created for the same sink URI in different routes distinct Kafka
transactional IDs. Configure an explicit, stable
+ `routeId` for each route, and choose a base `transactionalId` that is unique
to this Camel runtime instance and
+ stable across restarts. The base ID must still differ across application
replicas; the route suffix does not make two
+ different Camel instances globally unique. If the same route creates
multiple independent producer instances with the
+ same base ID, configure distinct base IDs for those producers as well.
+* Consumers reading the destination topic: `isolation.level=read_committed`.
+
+[source,java]
+----
+from("kafka:in?groupId=my-group&exactlyOnce=true&allowManualCommit=true&autoCommitEnable=false&breakOnFirstError=true")
+
.to("kafka:out?exactlyOnce=true&transactionalId={{orders.kafka.transactional-id}}");
+----
+
+The manual-commit token is used by Camel to capture source offset metadata; do
not call its `commit()` method in
+this mode. The consumer group metadata is captured on the Kafka polling thread
and retained in an exchange property,
+so `removeHeaders("*")` or replacing the message does not discard the source
offsets. The producer sends the offsets
+to the transaction with `sendOffsetsToTransaction` immediately before
`commitTransaction`. The metadata must belong
+to one consumer group generation; a rebalance before the transaction commits
causes Kafka to reject the stale
+generation and the transaction is recovered as a failure. Kafka brokers must
be version 2.5 or later to support the
+full `ConsumerGroupMetadata` fencing information.
+
+The source endpoint must use an explicit, non-blank `groupId`, must not
configure `offsetRepository`, and must use
+`consumersCount=1`. The built-in synchronous and asynchronous manual-commit
factories are supported. A custom
+`KafkaManualCommitFactory` is not supported because Camel needs the record
offset, leader epoch and poll-thread-captured
+group metadata. `batching=true`, `consumersCount` greater than one, `seekTo`,
and a `ResumeStrategy` are rejected on the
+source: batching can span consumer group generations, while seek/resume
policies can reposition the consumer past
+uncommitted records. Configure the source and destination endpoints against
the *same Kafka cluster*; cross-cluster
+offset commits are not supported. Streaming mode currently runs one source
record per transaction.
+For throughput, tune the Kafka producer's `batch.size` and `linger.ms`
options; do not enable consumer `batching`
+with exactly-once mode. To publish to several output topics in one
transaction, use one exactly-once producer endpoint
+and set the `CamelKafkaOverrideTopic` header for each output record rather
than routing to multiple Kafka producer
+endpoints.
+
+`breakOnFirstError=true` is required so a failed record is replayed from the
last committed offset. If the input topic
+is itself written transactionally, configure `isolation.level=read_committed`
on the source consumer as well.
+
+Exactly-once mode supports one Kafka source and one exactly-once producer
transaction for the normal source route.
+A Kafka producer used by the normal route must set `exactlyOnce=true`; a
non-transactional or ordinary transactional
+destination is rejected as a source-claim failure when the exchange is
processed, not during route startup. This is
+not a Kafka send failure: even if an error handler consumes the rejection, the
source transaction is aborted and the
+source record is replayed rather than skipped. A Kafka producer used as the
destination of the selected
+failure handler is supported as a separate case. This includes a
`deadLetterChannel` destination and
+an `onException(...).handled(true)` destination. Sends after a `doCatch`
block, as well as sends in the
+original route once processing resumes following an
`onException(...).continued(true)` clause, are
+normal route sends, not failure-handler deliveries. If a
Review Comment:
adressed
--
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]
