Repository navigation
Conversation
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
oscerd
left a comment
oscerd
left a comment
There was a problem hiding this comment.
Reviewed the exactly-once protocol, the failure handling and the configuration guarding closely — this is a serious, well-engineered feature, and the parts that are easy to get subtly wrong are handled correctly.
The transaction protocol is right. KafkaTransactionSynchronization sends the consumed offsets into the producer transaction with sendOffsetsToTransaction(offsets, ConsumerGroupMetadata) and only then commitTransaction() — the correct read-process-write ordering, and the test pins it with order.verify(...). On a failure of sendOffsetsToTransaction the test verifies abortTransaction() runs and commitTransaction() never does, so a mid-transaction failure aborts rather than partially commits. Tying this to a UoW Synchronization (commit on success, abort on failure), with care that a beginTransaction() failure doesn't leave a dangling synchronization, is the right structure, and using the ConsumerGroupMetadata overload (with the documented broker 2.5+ requirement) gets the zombie-fencing variant.
The prerequisite validation is what makes this safe in practice. validateExactlyOnceConfiguration() fails fast, before the consumer starts, unless every EOS precondition holds, and crucially it checks the effective consumer properties rather than just the typed option — so additionalProperties[enable.auto.commit]=true can't silently re-enable Kafka's background commits and break exactly-once behind the user's back. It also requires allowManualCommit=true, breakOnFirstError=true, an explicit groupId, and rejects a conflicting offsetRepository (offsets live in the transaction). That's the full set of foot-guns closed with clear messages, which is exactly what you want for a feature where a quiet misconfiguration silently degrades to at-least-once.
Config/compat. exactlyOnce and transacted default false and isolationLevel defaults read_uncommitted, so existing Kafka routes are unchanged — this is purely opt-in, which is why no upgrade-guide entry is needed. The new options are documented and the catalog/component/endpoint-DSL metadata is regenerated to match.
Scope note, not a blocker: this is a large change (~3k lines) and I focused on the EOS correctness core — the transaction lifecycle, the failure/abort path, and the validation. I did not line-by-line the commit-manager wiring (CommitManagers/AbstractCommitManager/NoopCommitManager), the per-partition highest-offset retention for batches, or the fencing/maxBlockMs edge behaviour; those are where a committer with Kafka depth and the EOS integration tests (CI was still running) should give it a final pass. But the protocol, the abort-on-failure guarantee and the configuration guardrails all check out.
Nice work.
Claude Code on behalf of oscerd
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The implementation is solid. Good use of the three-flag pattern (transactionStarted, unitOfWorkMarked, synchronizationRegistered) for atomic transaction initialization with correct exception suppression in all cleanup paths. The KafkaExactlyOnceMetadata immutability design and generation-consistency validation are correct. Integration tests cover abort-and-replay, batch offset survival across body transformation, and rebalance fencing — the right failure modes.
One minor duplication to clean up (see inline).
This review was generated by an AI agent, Hermès on behalf of @gnodet.
✅ Generated files are up to dateAn earlier CI run reported uncommitted generated changes; the latest run no longer does. |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — the dead-code finding from the previous SHA is still present.
The new commits correctly add the AtomicReference<String> activeExactlyOnceExchange concurrency guard and the matching releaseExactlyOnceTransaction cleanup. The three-flag commit protocol and exception-suppression paths look right.
One item not addressed (see inline).
This review was generated by an AI agent, Hermès on behalf of @gnodet.
|
🧪 CI tested the following changed modules:
🔁 2 tests passed only after a retry on JDK 21 (2 retried attempts) Recovered flaky tests on JDK 21 (2)
🔬 Scalpel shadow comparison — Scalpel: 11 of 713 tested, 24 compile-only — current: 11 all testedMaveniverse Scalpel detected 11 affected modules (current approach: 11). Skip-tests mode would test 11 modules (5 direct + 9 downstream), skip tests for 24 (generated code, meta-modules) Modules Scalpel would test (11)
Modules with tests skipped (24)
All tested modules (38 modules, 5m 24s total)Total reactor time: 5m 24s
Top 20 slowest modules:
|
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Previous findings are all addressed. LGTM.
The dead-code duplicate split/multicast guard in startKafkaTransaction (which was raised in my previous two reviews) has been removed in the latest commit. The canonical check in createExactlyOnceMetadata is the right place for this validation — close to the data it reads — and it now stands alone without the unreachable outer copy.
No new issues introduced by the removal.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
Croway
left a comment
Croway
left a comment
There was a problem hiding this comment.
Thanks for this feature. Before it ships I ran the exactly-once mode against real routes: a Kafka 4.4.0 broker (the test-infra container), kafka-clients 4.3.1 and PR head db8a96ff. I wrote one integration test per scenario. Each test asserts the faulty outcome, so a passing test means the problem reproduced. The numbers below are what the tests observed.
Several scenarios lose records silently, and several leave the route replaying the same record forever. Most of the cases come from three root causes:
- The consumer decides whether to move on from
exchange.getException()alone, but the transaction outcome is decided later, inonDone. A transaction can be aborted while the exchange has no exception (rollbackOnly), or fail after the error handler has handled the exception. In the first case the consumer moves on and the record is lost. In the second, it replays forever. - The source offsets and the "who commits them" decision travel with the exchange in fragile ways. They use a message header, a split/multicast property heuristic, and a synchronization that can be handed over to a copy with another exchange ID.
- Anything that repositions the consumer breaks the "replay from the committed offset" contract. That includes
seekTo, a resume strategy, and a batch dropped on error.
Most likely in production, in order
| # | Trigger | Effect | Reproduced |
|---|---|---|---|
| 1 | batching=true and a rebalance while a batch is filling (rolling deploy, scale-out, pod restart) |
Silent loss of every record in the batch | yes |
| 2 | filter/choice branch that does not send to Kafka, or exactlyOnce forgotten on the producer |
Offsets never committed; records reprocessed on every restart or rebalance; the whole topic if the producer flag is missing | yes |
| 3 | Poison record handled by deadLetterChannel / handled(true) |
Endless replay, one more DLQ copy per replay | yes |
| 4 | removeHeaders("*") (or anything that replaces the message) before the producer |
Endless replay with a misleading error | yes |
| 5 | .to("seda:...") after the producer (asynchronous hand-over) |
Producer slot leaked: route stuck until restart, nothing committed | yes |
| 6 | seekTo=END / seekTo=BEGINNING on the source |
END: silent loss. BEGINNING: committed duplicates | yes |
| 7 | split/multicast with an aggregation strategy before the producer |
Parent exchange rejected, endless replay | yes |
| 8 | markRollbackOnly() |
Silent loss of the record | yes |
1. Batch spanning a rebalance loses records (KafkaRecordBatchingProcessor.processBatch)
from("kafka:in?groupId=g&exactlyOnce=true&allowManualCommit=true&autoCommitEnable=false&breakOnFirstError=true"
+ "&batching=true&maxPollRecords=6&pollTimeoutMs=10000")
.to("kafka:out?transacted=true&exactlyOnce=true");- b0..b2 are polled and wait in the batch.
- A second member joins the group and the group rebalances.
- The consumer re-fetches b0..b2 under the new generation, which fills the batch.
KafkaExactlyOnceMetadata.fromExchangesthrows "exactlyOnce cannot combine records from different Kafka consumer group generations" before the route runs.processAndClearBatchclears the queue and the exception reaches the poll-exception strategy.SeekUtillogs "not advancing any partition offsets", but the position is already past b2.- b3 and b4 are produced and processed.
Result: polls [3, 3, 2], the route never saw b0..b2, committed output [b3, b4], committed offset 5. b0..b2 are lost.
A batch that fills over several polls spans any rebalance. That is normal under moderate traffic, and a rebalance happens on every rolling deploy. The cooperative assignor also bumps the generation, so it does not avoid this. Suggestion: never drop a batch on error in this mode. Return a break-on-error result so the consumer reconnects and rewinds to the committed offsets, and flush or discard the pending batch from the rebalance listener so a batch never spans two generations.
2. Offsets are only committed by an exactly-once producer transaction
from(eosSource).filter(header("kafka.KEY").startsWith("m")).to(eosSink); // m0, skip1, skip2, skip3All 4 records were processed, but the committed offset stayed at 1. After a route restart, skip1..skip3 were consumed again. With .to("kafka:out?transacted=true") (the exactlyOnce flag forgotten on the producer), no offset was ever committed and nothing warned about it. After a restart the output was [m0, m1, m2, m0, m1, m2].
Content-based routing is everywhere. The side effects of the non-Kafka branch (DB writes, HTTP calls) repeat after each restart or rebalance until a later Kafka-bound record covers the offset, and consumer lag stays high. Suggestion: when an exchange completes and no exactly-once transaction claimed its offsets, commit them from the consumer, which is safe because nothing was produced. Warn when a record from an exactly-once source goes through a Kafka producer without exactlyOnce.
3. Dead letter channel cannot skip a poison record (KafkaTransactionSynchronization.onDone)
errorHandler(deadLetterChannel("direct:dlq"));
from(eosSource).to("kafka:out?transacted=true&exactlyOnce=true&maxRequestSize=1024"); // small-0, big-1 (4 KB), small-2Within about 2 s, big-1 was attempted 4 times and reached the DLQ 3 times. Each commit failed with KafkaException: Cannot execute transactional method because we are in an error state, and small-2 was never reached. The loop does not end.
The cause: the send error puts the transaction into an abortable state, and the DLC then marks the exchange handled. onDone sees no exception and commits. The commit fails, the exception is set back on the exchange, and breakOnFirstError replays the record. A DLC is how users get past poison records (RecordTooLargeException, authorization on the output topic, invalid records). Suggestion: record the send failure in the synchronization. After a handled failure, abort, then commit the source offset alone in a new transaction, which skips the record atomically. At the very least, do not attempt a commit, and document the behaviour.
4. removeHeaders("*") breaks single-record mode (KafkaProducer.createExactlyOnceMetadata)
from(eosSource).removeHeaders("*").to(eosSink);Every attempt failed with "exactlyOnce requires a CamelKafkaManualCommit header; configure allowManualCommit=true on the Kafka source endpoint", although that option is set. m0 was replayed forever and nothing was committed. removeHeaders("*") is a common way to stop source headers leaking into the output record. pollEnrich with the default strategy, or a processor that sets a new Message, has the same effect. Suggestion: carry KafkaExactlyOnceMetadata in an exchange property in single-record mode too, as the batch path already does.
5. Asynchronous hand-over leaks the producer slot (KafkaTransactionSynchronization, KafkaProducer.releaseExactlyOnceTransaction)
from(eosSource).to(eosSink).to("seda:audit");SynchronizationAdapter.allowHandover() returns true, so SEDA moves the transaction synchronization to its copy. What happens next:
- The original exchange completes before the copy commits.
- The next record fails with "A Kafka exactlyOnce transaction is already active for exchange …-0000", which forces a reconnect.
- Because of the reconnect, the copy's delayed commit fails with "consumer group metadata mismatch: The coordinator is not aware of this member".
- The slot is then released with the copy's ID: "Could not release exactly-once producer slot for exchange …-0003; active exchange is …-0000".
Result: 51 "already active" failures in 28 s, nothing committed (not even m0), until the application restarts. Suggestion: override allowHandover() to return false, and release with the exchange ID stored when the slot was acquired.
6. seekTo is applied again on the reconnect that breakOnFirstError forces
SeekPolicyPartitionAssignmentAdapter seeks on every assignment. One failure on m2 gave:
seekTo=BEGINNING: committed output[m0, m1, m0, m1, m2, m3], so m0 and m1 were committed twice.seekTo=END: committed output[warmup, m0, m1, m4]. m2 (the failed record) and m3 were never consumed.
Suggestion: reject seekTo and a ResumeStrategy in validateExactlyOnceConfiguration, as is already done for offsetRepository.
7. Split/multicast check rejects the parent after an aggregation strategy
from(eosSource).split(body().tokenize(","), AggregationStrategies.useLatest()).transform(body().append("!")).end().to(eosSink);"exactlyOnce does not support split/multicast child exchanges" on the parent, replayed forever. MulticastProcessor.doDone → ExchangeHelper.copyResults copies the last sub-exchange's internal properties, including CamelSplitIndex, onto the parent. Checking the properties also misses copies that are not split children (see the wireTap case below). Suggestion: let the first transaction that includes the metadata mark it as claimed (compare-and-set on the metadata object kept in the exchange property), and reject any later claim.
8. markRollbackOnly() loses the record
from(eosSource).to(eosSink).filter(header("kafka.KEY").isEqualTo("m1")).markRollbackOnly().end();Committed output [m0, m2], committed offset 3, m1 attempted once. onDone aborts the transaction, including m1's offset, but the consumer only checks getException() and moves on. Suggestion: in exactly-once mode, treat a rolled-back transaction as a failure in the consumer. For example, the synchronization could record the outcome on the exchange, and the consumer would break and replay when it was aborted.
Lower probability
- wireTap to a second exactly-once producer. The copy commits the source offset in its own transaction. With a step after the main send that fails 1 s later, the output was
[m0], the audit output[m0, m1, m2]and the committed offset3, with m1 attempted once: m1's main output is lost. Without the delay, the second tap was rejected with "already active", because the previous tap still held the audit producer's slot. The claim in point 7 would cover this. consumersCount > 1. For 30 records on 3 partitions there were 3 "already active" rejections and 33 attempts. The output stayed exactly-once and drained in 6.8 s, so the cost is extra replays and rebalances. Rejecting it at startup, or documenting it, would avoid surprises.- Producer on another cluster. Every commit fails with "Specified group generation id is not valid" and the record is replayed forever (24 attempts in 12 s). When the input topic does not exist on the producer's cluster, each commit blocks for about 60 s first. Kafka does not support exactly-once across clusters, so a clear startup error would be friendlier.
Tests and smaller points
to.assertIsSatisfied(30000)/(60000)in the three new ITs do not extend the wait. That argument is only used when the mock expects no messages. A mock expecting one message that arrives after 15 s failsassertIsSatisfied(60000)after 10 003 ms. UseMockEndpoint.assertIsSatisfied(context, 60, TimeUnit.SECONDS).- In this mode
NoopCommitManager.commit()still logs at INFO "Auto commit … is enabled via Kafka consumer (NO-OP)", which is wrong here. It ran on every failure and stop: 182 lines in my run. - Fully qualified class names in
KafkaConsumer.validateExactlyOnceConfiguration(DefaultKafkaManualCommitFactory,DefaultKafkaManualAsyncCommitFactory) andKafkaProducer.createExactlyOnceMetadata(java.util.Collections, which is already imported).
I can share the reproduction test class (KafkaExactlyOnceReproIT, one test per scenario above) if it helps.
Claude Code on behalf of Croway
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
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:
- Default
transactional.idis not unique per instance (KafkaProducer.java:155-159). Withtransacted=trueit isgetEndpoint().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 eachSendProcessorcreates its own producer. This predates the PR, but the new doc recommendstransacted=truefor exactly-once. Please document that thetransactionalIdmust be unique per instance (e.g. from the pod/host name), or better, letexactlyOnce=truerequire an explicittransactionalIdor add a per-instance suffix. - No producer-side unit tests:
KafkaProducerTestis unchanged. ThedoStartcheck, the reject branches ofcreateExactlyOnceMetadata, the cleanup instartKafkaTransactionwhenbeginTransaction/addSynchronizationthrows, andreleaseExactlyOnceTransactionare 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 recommendingbatching=truefor throughput, and pointing to theCamelKafkaOverrideTopicheader 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
exactlyOnceoption description (KafkaConfigurationsetter javadoc) is what tooling shows; it should mention the requiredallowManualCommit,autoCommitEnable=false,breakOnFirstErrorandgroupId. - Nit:
KafkaExactlyOnceMetadata.EXCHANGE_PROPERTYis the class FQCN, andEXACTLY_ONCE_PRODUCER_PROPERTYis private inKafkaProducer; aCamel...constant inKafkaConstantswould match the module style. KafkaExactlyOnceBatchITexpects 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
The updated tests and documentation are included in the pushed commit. The full camel-kafka module test and integration-test build passed locally. |
Thanks for the detailed reproductions. I have pushed fixes in commit Here is how each finding is addressed:
I also addressed the additional observations:
The source and destination must use the same Kafka cluster. This is documented, and an offset-commit failure emits a warning to verify the cluster and consumer group generation. I have not added a proactive cluster-identity check during startup. I ran the full camel-kafka module test and integration-test build successfully after these changes. |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — all prior gnodet-bot findings are fully addressed. The dead-code duplicate guard from the previous two reviews has been removed.
This commit addresses all 10 issues raised by @Croway and @davsclaus. I traced the critical paths:
- Two-transaction DLC skip (
commitOffsetsAfterHandledSendFailure): abort old tx → recreate producer if needed → commit source-only tx. All failure modes (abort failure, offset-commit failure) set an exception on the exchange, causingcompleteExactlyOnceProcessingto detect the abort flag and force consumer replay. No silent data-loss path found. - Claim mechanism for split+aggregation: Camel's Splitter preserves the parent exchange ID on sub-exchanges, so the aggregated parent correctly passes the
sourceExchangeIdcheck. WireTap copies get new IDs and are correctly rejected. - Filter-branch offset commit:
completeExactlyOnceProcessing()callingmetadata.commitOffsets()when unclaimed is correct — a non-Kafka destination (log, direct, mock) intentionally takes this path, advancing the source offset on the polling thread. - SEDA handover:
allowHandover() = falseintegrates withDefaultUnitOfWork.handoverCompletions()which skips non-handoverable synchronizations. The synchronization stays on the original exchange and fires correctly.
One minor nit (see inline).
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks @ramu11, this is a thorough round. I went through the delta against my review and @Croway's:
exactlyOnce=truenow needs an explicittransactionalId.- Batching,
seekTo,ResumeStrategyandconsumersCount > 1are rejected at startup. - Filtered records commit their offsets from the consumer.
- The metadata lives in an exchange property, and a claim binds it to the source exchange.
- Exactly-once synchronizations no longer allow handover, and the slot is released with the id captured at start.
- A rollback-only or aborted transaction now replays the record.
- The producer-side unit tests and the regression IT cover the scenarios.
All of my earlier points and Croway's #1-#8 look addressed in the code.
One remaining way to lose a record:
- The send-failure marker survives a redelivery.
KafkaExactlyOnceMetadata.recordSendFailurekeeps the first failure inCamelKafkaExactlyOnceSendFailure, and nothing ever removes it. Suppose attempt 1 fails insidestartKafkaTransaction(for exampleinitTransactions()times out while the producer is recreated, KafkaProducer.java:627) and the error handler redelivers. Attempt 2 claims, sends and succeeds. InKafkaTransactionSynchronization.onDone(lines 89-97) the exchange has no exception but still has the marker, socommitOffsetsAfterHandledSendFailureaborts the good output and commits the source offset: the record is lost. Could you clear the marker when a new synchronization is registered (a failure before that point does not belong to the transaction), or better, keep the send-failure flag on theKafkaTransactionSynchronizationinstance instead of the exchange? A unit test with a failed first start and a successful redelivery would cover it.
Smaller:
recordSendFailurealso runs for producers that are not exactly-once (KafkaProducer.java:431 and :522, KafkaProducerCallBack.java:89), so every failed Kafka send now leaves aThrowableexchange property. Please only record it when the exchange carries the exactly-once metadata.- Doc: a Kafka destination that is not exactly-once is rejected at runtime, not at startup, so the record replays (and a DLC runs again) until the configuration is fixed. Worth saying.
- The batch-path exactly-once code (KafkaRecordBatchingProcessor.java:271-273,
KafkaExactlyOnceMetadata.fromExchanges) is unreachable now that batching is rejected. It can go. - The doc lists the rejected options twice (kafka-component.adoc around lines 598 and 601).
Claude Code on behalf of Claus Ibsen. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
Thanks for the review. I've addressed the comments in commit
The changes are pushed to the PR branch. Thanks! |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — all five of davsclaus's remaining points are addressed in commit 6d46b1d5b2fb.
- Stale send-failure marker ✅ —
SEND_FAILURE_PROPERTYis now cleared insidestartKafkaTransactionimmediately aftersynchronizationRegistered = true, before any sends. On a successful retry,onDonereads a clean state and takes the commit path. Lifecycle ordering is correct: register sync → clear old marker → send →onDonecommits. - Non-EOS producer paths ✅ —
recordSendFailurenow checksEXCHANGE_PROPERTYbefore recording the failure, so ordinary Kafka send failures no longer pollute non-exactly-once exchanges with a staleThrowableproperty. - Runtime destination validation ✅ — Documentation updated to clarify that a destination without
exactlyOnce=trueis rejected when the exchange is processed (not at startup), and that the source record replays until the configuration is fixed. - Unreachable batching path ✅ —
KafkaExactlyOnceMetadata.fromExchanges()removed, the deadif (configuration.isExactlyOnce())block inKafkaRecordBatchingProcessorremoved. Clean. - Duplicate rejected-options listing ✅ — Both doc copies trimmed; the redundant second copy of the startup-rejection list is gone.
exchange.setException(null)no-op ✅ — Removed; replaced with a comment documenting why the invariant holds (theonDoneguard ensures the exception is already null on entry tocommitOffsetsAfterHandledSendFailure).
Test coverage is sound: failedExactlyOnceBeginTransactionReleasesTheMetadataClaimAndProducerSlot now verifies both the failure path (marker set via setProperty) and the retry path (marker cleared via removeProperty). sendFailureIsNotRecordedForANonExactlyOnceExchange covers the new EOS-only guard. LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
TL;DR: Thanks @ramu11, every point from the last round is in, but one way to lose a record is left: a redelivery inside the already-open transaction (KafkaProducer.java:689 + KafkaTransactionSynchronization.java:93). 2 minor points.
All of these look good:
- The marker is cleared when a new synchronization registers.
recordSendFailurenow only records for exchanges with exactly-once metadata.- The runtime-rejection doc is in.
- The batch dead code is removed.
- The doc duplicate is removed.
- The
setException(null)nit is fixed.
Remaining record loss: redelivery in the same transaction. Clearing the marker at KafkaProducer.java:642 covers a failure before the synchronization registers. Here is the case it misses:
- Attempt 1 begins the transaction and registers the synchronization, then the send fails and the marker is set.
- The error handler redelivers.
uow.isTransactedBy(transactionId)is true, so we take the "Using existing kafka transaction" branch, and the marker stays. - Attempt 2 succeeds.
onDonesees no exception plus the marker. It aborts the good output and commits only the source offset, so the record is lost.
A retry in the same transaction can succeed: in kafka-clients 4.3.1 KafkaProducer.doSend, only an ApiException moves the transaction into the error state. A SerializationException (for example a schema-registry hiccup in the serializer) or a Camel-side createRecord failure leaves it usable.
Could we, in the existing-transaction branch, when the marker is present, abortTransaction() + beginTransaction() and clear the marker? That would also make redelivery work after an ApiException. Alternatively we could reject redelivery inside an exactly-once transaction. A unit test with a failed send, a successful redelivery (isTransactedBy=true) and then onDone expecting commitTransaction (no abort) would cover it.
Smaller:
- Doc (adoc ~612-614): with
.to(eosSink).to(nonEosKafka)and a DLC that handles the error, the marker set by the second producer sendsonDonedown the skip path. The record is skipped rather than replayed. Worth stating, or replaying in that case. - The new retry test only verifies
removePropertyon a mock. Asserting the commit-vs-skip outcome through the synchronization would make it stronger.
Claude Code on behalf of Claus Ibsen. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
Thanks for the follow-up review. Addressed the remaining points in commit
Validation passed: Kafka producer tests and all reported Maven builds completed successfully. Changes are committed and pushed to the PR branch. Thanks for the review! |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — all prior findings are addressed. The new commit (f9d92812) is clean.
The redelivery guard added in startKafkaTransaction is correct:
- Placement — correctly sits in the
elsebranch (transaction already active,uow.isTransactedBy(transactionId) == true), guarded byconfiguration.isExactlyOnce()to avoid touching non-EOS paths. - Rollback semantics —
exchange.setRollbackOnly(true)is set before theIllegalStateExceptionis thrown, soKafkaTransactionSynchronization.onDone()enters therollback()path even if an outer error handler swallows the exception. The test at lines 318–328 verifies this explicitly: withisRollbackOnly() = trueandgetException() = null,onDonecallsabortTransaction()and never reachessendOffsetsToTransactionorcommitTransaction. SEND_FAILURE_PROPERTYidempotency —recordSendFailure()already guards against overwriting an existing marker (exchange.getProperty(SEND_FAILURE_PROPERTY) == null), so thecatchblock in the asyncprocess()path cannot clobber the original failure with the redeliveryIllegalStateException.- Test coverage — the new test (
rejectsRedeliveryAfterExactlyOnceSendFailureWithinActiveTransaction) correctly stubsuow.isTransactedBy()to returnfalse, true(matching the first- and second-call sequence), stubsSEND_FAILURE_PROPERTYfor the secondprocess()invocation, and validates the fullonDoneabort path end-to-end. - Documentation — the doc update accurately describes the new constraint and aligns with the implementation.
LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
Croway
left a comment
Croway
left a comment
There was a problem hiding this comment.
Thanks @ramu11 for the thorough rework. I re-ran my scenarios against f9d92812 on the same setup (Kafka 4.4.0 broker, kafka-clients 4.3.1). This time every test asserts the correct outcome.
All eight points from my first review are fixed, and so are the lower-probability ones:
| Scenario | Result at f9d92812 |
|---|---|
markRollbackOnly() |
m1 replayed, output [m0, m1, m2] |
removeHeaders("*") |
output [m0, m1] |
| split with an aggregation strategy, then send | output [c!], one attempt |
.to("seda:...") after the producer |
output [m0, m1, m2], no "already active" |
| DLC without redelivery, poison record | output [small-0, small-2], DLQ received it once |
| filter branch | offsets committed, nothing consumed again after a restart |
producer without exactlyOnce |
rejected, the record replays, nothing committed |
seekTo=BEGINNING/END, batching=true, consumersCount=2 |
rejected at startup |
| wireTap next to a main send that then fails | m1 replayed, output complete |
The new skip, claim and direct-commit paths open three cases that I think matter in production.
1. A Kafka DLQ topic silently loses the poison record
errorHandler(deadLetterChannel("kafka:orders-dlq")); // also with ?exactlyOnce=true&transactionalId=dlq-1
from(eosSource).to(eosSink + "&maxRequestSize=1024"); // small-0, big-1 (4 KB), small-2For both DLQ endpoints, the committed offset was 3, the output [small-0, small-2], and the DLQ topic was empty:
- The DLQ producer is rejected. A plain endpoint fails
validateExactlyOnceDestination. An exactly-once endpoint fails with "Only one exactlyOnce Kafka producer is supported per exchange/unit of work". - The DLC swallows its own failure:
deadLetterHandleNewExceptiondefaults to true, and it logs "New exception occurred during processing by the DeadLetterChannel". - The exchange ends up handled, so
onDonetakes the skip path and logs "Skipped source offset(s) … after a handled Kafka send failure".
The record is gone, and three WARN lines are the only trace. A DLQ topic is the usual dead letter of a Kafka route, and the doc's advice ("configure a dead-letter route when the rejected payload must be retained") leads straight here.
Suggestion: write the DLQ record inside the offset-only transaction (abort, begin, send the DLQ record, sendOffsetsToTransaction, commit). That makes the skip and the DLQ write atomic. Failing that, skip only when the dead letter delivery succeeded and replay otherwise, and document that the DLQ must not be a Kafka endpoint.
2. An asynchronous hop to the exactly-once producer commits the offset without the output
from(eosSource).to("seda:out"); from("seda:out").to(eosSink);
from(eosSource).wireTap("direct:out").to("log:main"); from("direct:out").to(eosSink);In both cases the committed offset was 3 and the output was empty. The copy cannot claim the metadata, and the error stays in the secondary route:
- SEDA: "exactlyOnce requires a Kafka consumer as the route source"
- wireTap: "exactlyOnce source metadata cannot be used by a split, wireTap or copied exchange"
The original exchange completes without a claim, so completeExactlyOnceProcessing handles it like a filtered record and commits the offset with commitSync. The consumer cannot tell a filtered record from a record whose only output went to a rejected copy.
Suggestion: when a claim is rejected (copied exchange, or fromEndpoint not the Kafka source), mark the shared KafkaExactlyOnceMetadata as rejected, and have the consumer replay instead of commit when it sees the mark. That covers copies that run before the original completes. For slower asynchronous hops only the doc can help: say that the exactly-once producer must be reached synchronously by the source exchange, and that a SEDA, wireTap or threads() hop loses the record.
3. Split fan-out replays forever
from(eosSource).split(body().tokenize(",")).to(eosSink);Each child is rejected with "cannot be used by a split, wireTap or copied exchange". The children have new exchange IDs (…0001 to …0003 in the log), the parent fails and the record replays: 54 attempts in 30 s, nothing committed. The restriction is documented, but this is the most common way to write a one-to-many route, and it only fails at runtime. This pattern works (output [a!, b!, c!] in one transaction, one attempt), so it would be worth putting next to the restriction in the doc:
from(eosSource)
.split(body().tokenize(","), AggregationStrategies.groupedBody()).transform(body().append("!")).end()
.to(eosSink); // the List body is sent as three records in one transactionSmaller
- Two routes with the same sink URI share one
transactionalId. Their two producers fenced each other once (InvalidProducerEpochException), although all 20 records were delivered exactly once. The doc says the ID must be unique per producer instance, but reusing one endpoint URI in two routes is the natural way to write it. A duplicate-transactionalIdcheck per CamelContext, or a per-route suffix, would avoid the fencing. - DLC with
maximumRedeliveries(2). The redelivery is rejected, the DLC handles the exchange, and the record is skipped (DLQ once, offset3). The doc says instead that a rejected redelivery replays the source record. The behaviour is reasonable, but the doc should say that with a DLC the record goes to the DLQ after the first send failure, and thatmaximumRedeliveriesnever retries an exactly-once send.
I can share the test class (KafkaExactlyOnceReReviewIT, 17 tests) if it helps.
Claude Code on behalf of Croway
|
" I can share the test class (KafkaExactlyOnceReReviewIT, 17 tests) if it helps." |
|
@ramu11 here it is. Drop it into It uses the test-infra Kafka container. Each test creates its own CamelContext, topics, group and
|
|
Thanks @Croway for the review. I've pushed the updates in commit Changes made:
Validation: Ran: mvn -pl components/camel-kafka install \
-DskipTests=false \
-DskipITs=falseResult: BUILD SUCCESS Relevant test results:
The supplied review integration test was used for local validation and is not included in the commit. All changes are committed and pushed. Could you please take another look? |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The PR introduces a large and well-structured Kafka exactly-once feature. Two issues found.
Finding 1 — groupId mismatch throws wrong exception type
In createExactlyOnceMetadata(), when the source endpoint's groupId doesn't match the captured consumer group metadata, the code throws IllegalStateException instead of ClaimRejectedException. Every other rejection path in this method throws ClaimRejectedException, which is the correct type to signal a claim refusal.
The impact is behavioral: recordSendFailure() explicitly filters out ClaimRejectedException instances, so it does not record them as send failures. A plain IllegalStateException falls through that filter and gets recorded as a Kafka send failure instead. This causes KafkaTransactionSynchronization.onDone() to enter the offset-only commit path rather than the rollback-and-replay path — the wrong recovery behavior for a claim rejection.
The fix is one line: replace throw new IllegalStateException(reason) with throw new KafkaExactlyOnceMetadata.ClaimRejectedException(reason), consistent with every other rejection path in the same method.
Finding 2 — hasAsynchronousHandoff() doesn't detect dynamic routing constructs
containsAsynchronousNode() checks for wireTap, threads, and static to[seda:...] patterns, but misses dynamic routing constructs that can also hand off to an async endpoint: recipientList, routingSlip, and toD("seda:..."). The toD shortName is "toD", not "to", so the label check on line 216 won't match it. A route using toD("seda:my-queue") would pass the async-handoff check undetected, and the exactly-once claim could be incorrectly attempted on an async copy.
This is a lower-severity gap — the common cases are covered — but worth documenting at least (a @implNote or TODO in the method), or adding detection for the missing cases if they are expected to be used with exactly-once routes.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
TL;DR: Thanks @ramu11, the redelivery fix is right and well tested: a retry inside the open transaction is rejected, so no good output is aborted. One way to lose a record is still there: a non-Kafka failure handled by a DLC whose DLQ is a plain Kafka endpoint (KafkaProducer.java:733 + KafkaTransactionSynchronization.java:129). The new async-handoff check also makes filtered records replay forever in some routes. 4 smaller points.
Addressed:
- Redelivery in the open transaction: rejected and marked rollback-only. The new test drives
onDoneand asserts abort, no offsets, no commit. (Small thing: its comment says rollback-only wins "even if the error handler handles"; in Camel the handled path clears rollback-only, which is why the DLC skip described in the doc works.) - The doc for a handled non-exactly-once destination.
- The shallow retry test.
1. A non-Kafka failure plus a Kafka DLQ loses the record.
errorHandler(deadLetterChannel("kafka:dlq"));
from(eosSource).to(eosSink).process(e -> { throw new IllegalStateException("boom"); });- No Kafka send failed, so
isSecondaryKafkaDeadLetterSendis false. - The DLQ producer is rejected by
validateExactlyOnceDestination, which records the rejection as the first send failure (:733). - The DLC handles its own new exception (
deadLetterHandleNewExceptiondefaults to true). onDonetakes the offset-only skip path: eosSink's output is aborted, the offset is committed, and the DLQ topic is empty.
With an exactly-once Kafka DLQ the same route hits rejectClaim and replays forever, with the DLQ attempted on each pass.
Could the validation rejection stop counting as a send failure (mark the claim rejected instead), and could a Kafka DLQ be allowed for any handled failure, not only after a Kafka send failure? An IT with this route would cover it.
2. hasAsynchronousHandoff scans the whole route model (KafkaRecordStreamingProcessor.java:183-213). getChildren() includes branches the exchange never took. So any route with a wireTap, threads() or to("seda:...") anywhere fails every record that no producer claimed. Example:
from(eosSource).filter(p).to(eosSink).end().wireTap("direct:audit");Here the filtered records replay forever and the consumer is stuck, while the doc says filtered records commit their offset. The check also misses toD/recipientList/routingSlip, and it walks the route tree for every filtered record. Could we rely on rejectCopyClaim and the TO_ENDPOINT check plus the doc, or only look at the path the exchange actually took? A filter + wireTap test would pin it.
Smaller:
org.apache.camel.Routeis fully qualified atKafkaRecordStreamingProcessor.java:192; please import it.SECONDARY_SEND_FAILURE_PROPERTY/SUPPRESS_NEXT_SEND_FAILURE_PROPERTYcould live inKafkaConstantslike the other exactly-once properties.- The ITs cover a failed Kafka DLQ but not a successful one (record in the DLQ, offset skipped). Worth adding, since the doc recommends it.
- groupId mismatch (
KafkaProducer.java:802): throwingClaimRejectedExceptionwould match the other rejection paths. TodayonDonestill rolls back because it checksisClaimRejected()first.
Claude Code on behalf of Claus Ibsen. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — both prior gnodet-bot findings are addressed in commit a5700ed2. Two of davsclaus's open points remain unresolved.
Prior findings — status:
- ✅
groupIdmismatch now throwsClaimRejectedException(line 802) instead ofIllegalStateException— consistent with every other rejection path. Test updated withassertNull(metadata.getSendFailure(), ...)verifying the rejection is not recorded as a Kafka send failure. - ✅
containsAsynchronousNodenow detectsrecipientList,routingSlip, andtoD[seda:...]. Unit tests added for each.
Remaining issues:
See inline comments for details on the two open correctness bugs from davsclaus's last round.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
| IllegalStateException failure = new IllegalStateException( | ||
| "A Kafka exactlyOnce source requires every Kafka destination producer to set exactlyOnce=true " | ||
| + "and configure an explicit transactionalId"); | ||
| KafkaTransactionSynchronization.recordSendFailure(exchange, failure); |
There was a problem hiding this comment.
validateExactlyOnceDestination records its own rejection as a Kafka send failure via recordSendFailure. This enables a silent record-loss path:
- Processing fails (non-Kafka exception, e.g. an
IllegalStateExceptionfrom aprocess()step). - DLC routes the exchange to
kafka:dlq(a plain Kafka endpoint, noexactlyOnce=true). validateExactlyOnceDestinationfires:isSecondaryKafkaDeadLetterSend()returns false (no prior Kafka send failure), so it callsrecordSendFailureand throwsIllegalStateException.- DLC handles its own new exception (
deadLetterHandleNewException=true) —exchange.getException()is cleared. KafkaTransactionSynchronization.onDone():sendFailure != null, no exception → takescommitOffsetsAfterHandledSendFailure→ aborts the (successfully sent)eosSinkoutput and commits only source offsets. DLQ topic is empty.
The record is silently lost. The offset-skip path was designed for a genuine Kafka send failure handled by the error handler. A DLQ configuration rejection is not a Kafka send failure. Fix: remove the recordSendFailure call from validateExactlyOnceDestination and instead mark the exactlyOnceMetadata claim as rejected (analogous to the existing rejectClaim pattern) so onDone takes the rollback path rather than the offset-skip path.
| if (routeId == null) { | ||
| return false; | ||
| } | ||
| org.apache.camel.Route route = exchange.getContext().getRoute(routeId); |
There was a problem hiding this comment.
containsAsynchronousNode(route.getRoute()) walks the entire route tree, including branches the exchange never took. For example:
from(eosSource).filter(predicate).to(eosSink).end().wireTap("direct:audit");A filtered record (bypasses .to(eosSink)) reaches completeExactlyOnceProcessing with !metadata.isClaimed(). hasAsynchronousHandoff walks the whole route → finds wireTap → returns true → sets an exception → consumer replays the record forever.
With this commit, recipientList and routingSlip are now also unconditionally flagged as async. Any route containing those EIPs — regardless of whether the filtered record ever reached them — causes filtered records to replay forever. This is a regression for the filter + recipientList pattern.
The TO_ENDPOINT check at lines 184–186 is already the right approach (it reflects what the exchange actually did). The static route-tree walk is not. Fix options:
- Remove the tree walk entirely — rely on
TO_ENDPOINTforseda:detection and add a flag that async dispatchers (wireTap, threads) set on the exchange when they actually fire. - Only walk the route tree when
TO_ENDPOINTis not set and limit detection to direct children of the route root (not deep children in branches).
Also: org.apache.camel.Route is fully qualified here; please add import org.apache.camel.Route; and use the simple name.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
TL;DR: Thanks @ramu11. a5700ed lands 4 minutes after my previous comment and applies gnodet-bot's suggestions; the groupId mismatch now throwing ClaimRejectedException is good. My previous comment still stands in full, and one change here makes its point 2 wider.
containsAsynchronousNode (KafkaRecordStreamingProcessor.java:196-219) now also flags recipientList, routingSlip and toD[seda:. The scan still walks the whole route model (getChildren() includes branches the exchange never took), so with this change any exactly-once route that has a recipientList or routingSlip anywhere fails, and replays forever, every record that no producer claimed (for example the records a filter drops). The new tests mock NamedNode, so they check the detection but not this false positive. Basing the decision on what the exchange actually did (TO_ENDPOINT, or a marker set when an async EIP actually runs) would fix both the old and the new cases; a filter + wireTap or filter + recipientList route test would pin it.
The other points of my previous comment are still open: the record loss with a Kafka DLQ (KafkaProducer.java:733), the FQCN at :192, the constant placement and the happy-path DLQ IT.
Claude Code on behalf of Claus Ibsen. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
Description
This PR adds opt-in read-process-write exactly-once processing support to the Apache Camel Kafka component.
When enabled, Kafka output records and the corresponding source consumer offsets are committed atomically in a single Kafka transaction. This prevents output records from becoming visible when processing fails before the source offsets are committed, and allows failed input records to be replayed safely.
Implementation
exactlyOnceoption to the Kafka component configuration and exposes it through generated component and endpoint DSLs.sendOffsetsToTransaction()API to include source offsets in the output transaction.Configuration requirements and limitations
For the Kafka source endpoint, configure:
exactlyOnce=trueallowManualCommit=trueautoCommitEnable=falsebreakOnFirstError=truegroupIdFor the destination Kafka producer, configure
exactlyOnce=trueand enable transactions usingtransacted=trueor atransactionalId.Consumers reading the transactional output should use
isolation.level=read_committed.Kafka brokers version 2.5 or later are required. The current implementation supports one Kafka source and one exactly-once transactional producer per exchange/unit of work. Using an
offsetRepositoryor a customKafkaManualCommitFactoryis not supported. Split/multicast child exchanges are not supported in exactly-once mode.Tests
Adds unit and integration coverage for:
Existing Kafka routes remain unchanged unless
exactlyOnceis explicitly enabled.