allthingssecurity opened a new pull request, #29287: URL: https://github.com/apache/camel/pull/29287
# Description [CAMEL-26598](https://issues.apache.org/jira/browse/CAMEL-26598) With `breakOnFirstError=true`, `KafkaFetchRecords` skipped the break when the consumer was paused in the same poll loop iteration (`result.isBreakOnErrorHit() && !state.equals(PAUSED)`). The consumer was kept, with its position after the polled records, and on resume only the partitions with a committed offset were seeked. On a partition without a committed offset (a new group or partition, the first records of a group) and `autoOffsetReset=earliest`, the failed record and the rest of its poll were never routed, also with `exactlyOnce=true`, whose docs say `breakOnFirstError=true` is required so a failed record is replayed. (With `autoOffsetReset=latest`, the default, a break on such a partition skips to its end with or without a pause; not changed here.) No race is needed: `ThrottlingExceptionRoutePolicy` suspends the consumer from `onExchangeDone` of the failed exchange, on the fetcher thread, so the pause and the break always come in the same iteration. Found with a TLA+ model of the consumer whose traces were checked against the real consumer (driven by `MockConsumer`). Change (`KafkaFetchRecords`): - `startPolling`: the break creates a new Kafka consumer also when the consumer is paused. The `!PAUSED` condition came from CAMEL-18588, when a new Kafka consumer was not paused again, so the break would have resumed a suspended route. Since CAMEL-25313 a suspended consumer keeps its fetcher thread and the rebalance listener pauses the partitions assigned to a new Kafka consumer, so it polls from the committed offset, or from `autoOffsetReset` without one, once the route is resumed: the same as a break without a pause. The resume seek of CAMEL-18327 stays. - `run`: a new Kafka consumer of a suspended consumer also pauses the partitions it already has, assigned without a rebalance (`assign()` in a custom `SubscribeAdapter`), before its first poll. Without this, the break above made such a suspended route route the records again (and with a failing record, reconnect in a loop). Cost and alternative: each such break makes the consumer leave and rejoin the group (a rebalance), once for each failed exchange that suspends the consumer, also on a partition with a committed offset, where the resume seek already replayed the record. Keeping the paused consumer and seeking instead would have to move back every partition with unprocessed records of the poll (the break also skips the other partitions' records), and for a partition without a committed offset it would have to re-implement each `autoOffsetReset` strategy (earliest, latest, by_duration) per partition. The reconnect is the path a break without a pause already takes, and one code path for the break is easier to reason about than a paused variant (only reconnecting when a partition has no committed offset would keep the CAMEL-18327 resume seek as a second replay path for the same failure, plus a `committed()` call on each break). A circuit breaker suspends the consumer at most once per `halfOpenAfter`, s o the extra rebalances are bounded by that. Upgrade guide (4.24) entry added; there is no option to keep the previous behaviour. Main only (4.24). Not covered: `autoCommitEnable=true`, where the Kafka client commits positions by itself. Tests (`KafkaConsumerBreakOnFirstErrorPausedTest`, no broker: a `MockConsumer` per Kafka consumer polls a simulated log from the committed offsets of the group): - Fail on main (the records are never routed again): circuit breaker with synchronous manual commit; circuit breaker with `exactlyOnce` (also checks the output records and the committed source offset); a suspend requested while the failed record is in the route, with subscribed and with assigned partitions (the assigned case also fails without the pause in `run`: the records are routed while suspended). - Pass on main and with the change: a break without a pause; a circuit breaker on a partition with a committed offset (replayed by the resume seek). - On main 4 of the 6 tests fail; with the change 6/6. The new test with `KafkaConsumerSuspendResumeTest`, 20x: 240/240. - camel-kafka unit tests (offline, `-DskipITs`): 283 run, 0 failures, 1 skipped. # 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-26598](https://issues.apache.org/jira/browse/CAMEL-26598). # 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 camel-kafka and the modules it depends on, including the formatter and import-sort plugins, and ran the camel-kafka unit tests. No generated file changes. I did not run the full root build.) # 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]
