ramu11 commented on PR #28730: URL: https://github.com/apache/camel/pull/28730#issuecomment-6539125982
> Thanks for taking on exactly-once support, it is a big and welcome piece of work. > > I agree with @Croway's review: I traced #1 (a generation mismatch drops the batch: `KafkaRecordBatchingProcessor` throws in `fromExchanges` and `processAndClearBatch` clears the queue in `finally`), #3 (an exchange marked handled by the error handler/DLC still commits the transaction that is in error state, so the record replays forever) and #8 (`markRollbackOnly()` aborts the transaction, but the consumer only checks `getException()` and moves on, so the next transaction commits past the record) in the code, and they behave as described. Losing records is the one thing this feature must not do, so these need a fix (or a refusal at startup) before merge. > > Two more points that are not raised yet: > > 1. **Default `transactional.id` is not unique per instance** (`KafkaProducer.java:155-159`). With `transacted=true` it is `getEndpoint().getId() + "-" + routeId`, and the endpoint id is `"endpoint" + counter` (`EndpointHelper.createEndpointId`). Two replicas of the same application (the normal way to scale a consumer group) get the same id and keep fencing each other (`ProducerFencedException`, recreate, fence the other one ...). Two `.to(sameUri)` in one route have the same problem, as each `SendProcessor` creates its own producer. This predates the PR, but the new doc recommends `transacted=true` for exactly-once. Please document that the `transactionalId` must be unique per instance (e.g. from the pod/host name), or better, let `exactlyOnce=true` require an explicit `transactionalId` or add a per-instance suffix. > 2. **No producer-side unit tests**: `KafkaProducerTest` is unchanged. The `doStart` check, the reject branches of `createExactlyOnceMetadata`, the cleanup in `startKafkaTransaction` when `beginTransaction`/`addSynchronization` throws, and `releaseExactlyOnceTransaction` are not covered. On the IT side, a failing batch (abort, then the whole batch replays) is not covered either, and that is where #1 lives. > > Smaller: > > * Doc: streaming mode is one transaction per record (begin, `sendOffsetsToTransaction`, commit per record). Worth recommending `batching=true` for throughput, and pointing to the `CamelKafkaOverrideTopic` header on one producer endpoint to write several topics in one transaction, as the "one exactlyOnce producer per exchange" rule rejects fan-out to several endpoints. > * The `exactlyOnce` option description (`KafkaConfiguration` setter javadoc) is what tooling shows; it should mention the required `allowManualCommit`, `autoCommitEnable=false`, `breakOnFirstError` and `groupId`. > * Nit: `KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY` is the class FQCN, and `EXACTLY_ONCE_PRODUCER_PROPERTY` is private in `KafkaProducer`; a `Camel...` constant in `KafkaConstants` would match the module style. > * `KafkaExactlyOnceBatchIT` expects exactly one batch for 8 records, which may be flaky if they are polled in two batches. > > _Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying._ Thanks @davsclaus for the thorough review and for tracing the failure cases. I have pushed the changes in commit `e7671c84f2d7`. - **Transactional ID:** `exactlyOnce=true` now requires an explicit `transactionalId`. The documentation explains that it must be unique per producer instance and stable across restarts. `transacted=true` alone is not sufficient for exactly-once mode. - **Producer-side tests:** `KafkaProducerTest` now covers the exactly-once startup requirement, invalid destination/source configurations, missing or mismatched source metadata, transaction-start failure cleanup, synchronization-registration failure cleanup, and release of the producer slot using the original exchange ID. - **Batch handling:** Rather than supporting batch transactions with unsafe rebalance behavior, exactly-once mode now rejects `batching=true` at source startup. I removed `KafkaExactlyOnceBatchIT`, which assumed one batch for a fixed number of records. A broker-backed regression test verifies that batching is rejected. - **Throughput and multiple topics:** The documentation now recommends tuning producer `batch.size` and `linger.ms` while keeping consumer batching disabled. It also documents using one producer with `CamelKafkaOverrideTopic` to write to multiple output topics in one transaction. - **Option documentation:** The `exactlyOnce` setter javadoc now lists the required source options and explicit destination transactional ID. - **Exchange properties:** The exactly-once metadata and transaction markers now use shared `CamelKafka...` constants in `KafkaConstants`. - **Other review points:** The integration-test timeout assertions use the static `MockEndpoint` helper, and the duplicate split/multicast guard was removed. The updated tests and documentation are included in the pushed commit. The full camel-kafka module test and integration-test build passed locally. -- 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]
