allthingssecurity opened a new pull request, #29667: URL: https://github.com/apache/camel/pull/29667
# Description [CAMEL-26856](https://issues.apache.org/jira/browse/CAMEL-26856) With `exactlyOnce` on a Kafka consumer, once a Kafka send failed in the open transaction, `KafkaProducer` records the send failure and rejects the next send through the source producer: it marks the exchange rollback only, sets the `CamelKafkaExactlyOnceRetryRollback` property and throws. That next send is a redelivery of the failed send by an error handler (`maximumRedeliveries` > 0), or any other send through the producer, such as a second send in a `loop` or in another `doTry` block. `KafkaTransactionSynchronization.onDone` took the offsets-only path for such a rejected send (abort the output transaction, commit the source offsets alone, so the record is skipped) only when the exchange was failure handled, that is when an error handler handled the failure. A `doCatch` that catches the rejected send leaves no exception, but: - when the `doTry` failed in the route of the send, the exchange is not failure handled, so the rejected send was rolled back and the record replayed; - when the `doTry` called the route of the send (for example with `direct`), the exchange is still failure handled on main and took the offsets-only path; CAMEL-26447 (#29584, open) stops marking it, so after that PR this case is replayed too, which is what the JIRA reports. For a send that keeps failing (a record that is too large, for example) every replay ends the same way, so the consumer does not get past the record. Change: a rejected send counts as handled when the exchange has no exception left and the first send failure is recorded (or the exchange is failure handled), whoever handled it, an error handler or a `doCatch`. Then the recorded send failure takes the offsets-only path, as for a send failure that a `doCatch` catches without a rejected retry, and as for an error handler. - A rejection that nothing handled keeps its exception and is still rolled back (the branch on `exchange.getException() != null` comes first). - A rollback the route asks for without the marker (`markRollbackOnly`) is still rolled back, and so is the marker without a recorded send failure (not reachable today; a guard). - A `markRollbackOnly()` after a handled rejection is not told apart from the rejection; that was already so when an error handler handled it. The way to replay the record is to throw the exception again in the `doCatch` block, or `rollback()`, which set an exception. Visible change: a `doCatch` that catches a rejected send in the route of the send now skips the record, where it was replayed before; the 4.24 upgrade guide says so, with the way back. For the `doCatch` of a calling route the outcome stays the one of 4.23 once #29584 merges. This PR can be merged before or after #29584: both add to the 4.24 upgrade guide, in different sections, and merge cleanly. Docs: the Kafka component page says which sends are rejected, that a `doCatch` that catches one skips the record, and how to replay it instead; it also corrects its example of `maximumRedeliveries(2)`, which said that both retries are rejected: the first rejected retry marks the exchange rollback only, which ends the redeliveries (`RedeliveryErrorHandler` counts a rollback-only exchange as exhausted). The catalog copy is updated. Not changed (follow-up): after a `doCatch`, `CatchProcessor` sets the rollback-only mark again, so the consumer's completion (`KafkaRecordStreamingProcessor`) still sees a rollback-only exchange after the offsets-only commit and, with `breakOnFirstError`, logs that the record must be replayed and reconnects; the reconnect resumes from the committed offset, so the record is skipped, but with a misleading WARN and a group rejoin. This is so on main already for the `doCatch` of a calling route; for a `doCatch` in the route of the send it comes with the skip. I will file a JIRA for it. Tests, in `KafkaTransactionSynchronizationTest` (no broker, mocked producer and exchange as the other cases of the class; the source metadata holds the recorded send failure, as `KafkaProducer` leaves it): - `redeliveryRejectionCaughtByDoCatchOfCallingRouteCommitsOffsetsAfterAbortingOutput`: no exception, rollback only, the marker, not failure handled (the state a `doCatch` leaves in both cases above): abort, then a new transaction with the source offsets only, committed; the transaction is not marked aborted for the consumer. - `redeliveryRejectionThatNothingHandledRollsBack`, `rollbackOnlyWithoutRetryMarkerAfterDoCatchRollsBack` and `redeliveryRejectionMarkerWithoutRecordedSendFailureRollsBack`: controls, all rolled back (they pass before the change too). Without the change the first fails (the transaction is aborted and nothing is committed). With it the class passes (25 tests), and the camel-kafka unit tests pass (325 run, 0 failures, 1 skipped). A case in `KafkaExactlyOnceIT` (needs a Kafka container) is not added; I could not run it locally. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-26856](https://issues.apache.org/jira/browse/CAMEL-26856). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built the core modules and camel-kafka with the formatter and import-sort plugins and ran the camel-kafka unit tests; the catalog copy of the component page matches it. I did not run the full root build. The ITs, which need a Kafka container, were not run locally; CI runs them.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
