ramu11 commented on PR #28730:
URL: https://github.com/apache/camel/pull/28730#issuecomment-6539141373
> 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:
>
> 1. **The consumer decides whether to move on from
`exchange.getException()` alone, but the transaction outcome is decided later,
in `onDone`.** 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.
> 2. **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.
> 3. **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`)
> ```java
>
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");
> ```
>
> 1. b0..b2 are polled and wait in the batch.
> 2. A second member joins the group and the group rebalances.
> 3. The consumer re-fetches b0..b2 under the new generation, which fills
the batch.
> 4. `KafkaExactlyOnceMetadata.fromExchanges` throws "exactlyOnce cannot
combine records from different Kafka consumer group generations" before the
route runs.
> 5. `processAndClearBatch` clears the queue and the exception reaches the
poll-exception strategy. `SeekUtil` logs "not advancing any partition offsets",
but the position is already past b2.
> 6. 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
> ```java
> from(eosSource).filter(header("kafka.KEY").startsWith("m")).to(eosSink);
// m0, skip1, skip2, skip3
> ```
>
> All 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`)
> ```java
> errorHandler(deadLetterChannel("direct:dlq"));
>
from(eosSource).to("kafka:out?transacted=true&exactlyOnce=true&maxRequestSize=1024");
// small-0, big-1 (4 KB), small-2
> ```
>
> Within 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`)
> ```java
> 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`)
> ```java
> from(eosSource).to(eosSink).to("seda:audit");
> ```
>
> `SynchronizationAdapter.allowHandover()` returns `true`, so SEDA moves the
transaction synchronization to its copy. What happens next:
>
> 1. The original exchange completes before the copy commits.
> 2. The next record fails with "A Kafka exactlyOnce transaction is already
active for exchange …-0000", which forces a reconnect.
> 3. Because of the reconnect, the copy's delayed commit fails with
"consumer group metadata mismatch: The coordinator is not aware of this member".
> 4. 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
> ```java
> 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
> ```java
>
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 offset `3`, 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 fails
`assertIsSatisfied(60000)` after 10 003 ms. Use
`MockEndpoint.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`) and
`KafkaProducer.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_
Thanks for the detailed reproductions. I have pushed fixes in commit
`e7671c84f2d7` and added broker-backed regression coverage for the reported
scenarios.
Here is how each finding is addressed:
1. **Batch/rebalance data loss:** Exactly-once mode now rejects
`batching=true` during source consumer startup. Rather than risk losing a batch
across consumer group generations, this implementation currently supports
streaming mode only. A real-route startup regression test verifies the
rejection.
2. **Offsets for filtered records and non-EOS destinations:** When a
successful route does not use an exactly-once Kafka producer, the source
offsets are committed directly on the consumer polling thread. A Kafka
destination without `exactlyOnce=true` is rejected, leaving the source record
available for replay. Regression tests cover both filtered records and a
non-EOS Kafka destination.
3. **Handled send failures/DLC:** Kafka send failures are retained even when
the Camel error handler handles the exception. The failed output transaction is
aborted, and the source offsets are committed in a new offset-only transaction.
A broker-backed poison-record/DLC test verifies that the record is not replayed
indefinitely. External DLQ side effects are still outside the Kafka transaction
and are subject to the usual crash-boundary caveat.
4. **Message header removal:** Source offsets and consumer group metadata
are retained in an exchange property, so `removeHeaders("*")` or message
replacement does not lose them. There is a regression test for header removal.
5. **SEDA handover:** Exactly-once synchronization does not allow handover
to a copied exchange. The producer slot is released using the original exchange
ID captured when the transaction started. A SEDA regression test covers this
case.
6. **Seek/reconnect behavior:** `seekTo` and `ResumeStrategy` are rejected
for exactly-once sources so replay always follows committed offsets.
7. **Split/multicast and wireTap:** Source metadata can only be claimed by
the original source exchange. Split/multicast aggregation back to the parent is
supported when the Kafka producer is invoked after aggregation, but child and
wireTap copies cannot claim the source offsets. Regression tests cover both
cases.
8. **Rollback-only:** An aborted transaction is now explicitly marked on the
exchange and propagated to the streaming consumer, which treats it as a failure
and replays the source record. There is a broker-backed `markRollbackOnly()`
regression test.
I also addressed the additional observations:
- `consumersCount` greater than one is rejected at startup.
- Integration tests use the static `MockEndpoint.assertIsSatisfied(context,
timeout, unit)` overload.
- The exactly-once `NoopCommitManager` messages are logged at DEBUG.
- The fully qualified class references were cleaned up.
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.
--
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]
