allthingssecurity opened a new pull request, #28071: URL: https://github.com/apache/camel/pull/28071
# Description [CAMEL-25756](https://issues.apache.org/jira/browse/CAMEL-25756) With `batching=true` the Kafka consumer keeps the records of several polls in `KafkaRecordBatchingProcessor` until the batch is complete (`maxPollRecords` records), the poll timeout expires or `batchingIntervalMs` passes. `KafkaFetchRecords.startPolling()` created a new record processor facade, and so a new, empty batch, every time it was entered. It is entered again on the same connected Kafka consumer after every exception of the polling loop that the poll exception strategy continues on: a failed `poll` with the default `pollOnError=ERROR_HANDLER`, and with `DISCARD` or `RETRY`. The records held for the incomplete batch were then dropped. The Kafka consumer does not return them again (its position is past them), and the next complete batch commits its highest offset per partition, past them. Once a later batch of the same partition was committed, they were never routed and never redelivered, even after a restart, and nothing was logged about them. (If no later batch of that partition was committed before a stop or a revocation, they were polled again from the committed offset.) A realistic trigger is a record that a failing deserializer (Avro, JSON schema, a custom one) cannot read: `poll` throws `RecordDeserializationException`, the default error handler skips that record, and the good records polled before it (on every partition) were lost with it. Up to `maxPollRecords - 1` records per failed poll. This change creates the facade when the Kafka consumer is connected or reconnected, and keeps it while the consumer is connected. A failed poll keeps the held records; they are routed and committed with the next batch. After a reconnect the new Kafka consumer starts from the committed offsets, so a new facade, with an empty batch, is still created then. The streaming facade holds no state between polls and behaves as before. `KafkaFetchRecords` is the only production change. Not changed: - `breakOnFirstError` in batching mode still commits the failed batch before the reconnect (as `KafkaBatchingProcessingBreakOnFirstErrorIT` expects); - the records held for an incomplete batch are still dropped when the route stops; they were not committed, so they are polled again after the restart. Tests: - New `KafkaBatchingConsumerPollErrorTest` (2 tests, `MockConsumer`, Awaitility): a real route with `batching=true&maxPollRecords=10` and Kafka's `MockConsumer`. Three records are polled and held, then a poll fails with a `KafkaException` (first test) or with a `RecordDeserializationException` for offset 3 (second test), then ten more records arrive. Each test checks the records of the first batch and the committed offset. - Without the change (main's `KafkaFetchRecords`, same tests) both fail: `expected: <[a0, a1, a2, b0, b1, b2, b3, b4, b5, b6]> but was: <[b0, b1, b2, b3, b4, b5, b6, b7, b8, b9]>`. - With the change, the camel-kafka unit tests pass: 222 run, 0 failures, 1 skipped. The integration tests need a Kafka broker and were not run. How it was found: deterministic simulation testing with a fake Kafka consumer written from the KafkaConsumer Javadoc. The seed picked the consumer configuration (streaming or batching, commit managers, `breakOnFirstError`, `maxPollRecords`, timeouts), records that cannot be deserialized, failed polls and commits, and route restarts and suspends. The invariant "every record sent reaches the route" failed only with `batching=true` and a failed poll on main; with this change those seeds pass. The simulation harness is not part of this PR. # 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). # 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 and tested camel-kafka, including the formatter and import-sort plugins. 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]
