allthingssecurity opened a new pull request, #29288: URL: https://github.com/apache/camel/pull/29288
# Description [CAMEL-26599](https://issues.apache.org/jira/browse/CAMEL-26599) With `useKclConsumers=true`, a record whose exchange failed was checkpointed and never consumed again. CAMEL-24191 meant to prevent this, but: - a route that fails sets the exception on the exchange and returns normally, so `processRecords` reached `checkpointer.checkpoint()` right after the failure (the CAMEL-24191 test mocks a processor that throws, which a route does not do); - even when the checkpoint was skipped, KCL goes on with the next batch while the consumer holds the lease of the shard (it does not deliver a batch again), and `checkpoint()` covers every record delivered so far, so the next batch, or `shutdownRequested` when the route stops, checkpointed past the failed records. Found with a TLA+ model of the KCL record processor, compared with traces of the real record processor behind a real route (KCL's shard consumer emulated in-JVM). As discussed on the JIRA, today's default stays (a failed record is not consumed again, as with the plain `Kinesis2Consumer` and camel-kafka without `breakOnFirstError`), and at-least-once becomes an opt-in endpoint option. Change (`KclKinesis2Consumer.CamelKinesisRecordProcessor`, `Kinesis2Configuration`): - new option `kclStopCheckpointOnFailure` (consumer, advanced, default `false`). When enabled, once a record has failed (the exchange ends with an exception, or the processor throws), this record processor (one per shard lease) does not move the checkpoint past it: it checkpoints only the records of the batch before it (`checkpoint(sequenceNumber, subSequenceNumber)`), and later batches and `shutdownRequested` do not checkpoint. The records from the failed one on are consumed again when the lease is taken again (the route is started again, or the lease moves to another worker). One WARN per lease with the sequence number of the first failed record. The consumer keeps routing the shard, so the records after the failed one can be routed twice and out of order; this cost, and the replay of a record that always fails (up to the stream retention), are written in the option description and in the component docs. `shardEnded` still checkpoints the end of the shard, as KCL needs that to g o on with the child shards, with a WARN when a record of the shard failed; - by default (option off), an exchange that ends with an exception is now reported to the exception handler of the consumer, as the pubsub, pulsar and plain Kinesis consumers do, and the batch is checkpointed as before. A processor that throws still skips the checkpoint of that batch, as before; - either way, the checkpoint catches KCL's retryable exceptions (`ThrottlingException`, `KinesisClientLibDependencyException`) with a WARN without stack trace: the next checkpoint covers the records. Docs: a "Failed records with the KCL consumer" section in the component docs (default: not consumed again, use a dead letter channel or redelivery; with the option: the checkpoint stays before the failed record, and its cost). Upgrade guide (4.24): a short entry, the default is unchanged except the exception handler call; it names the new option. Generated files (configurers, URI factory, component JSON, catalog copy, endpoint and component DSL) regenerated for the new option. With the default kept, this can also go to 4.18.x and 4.14.x: both have the same `processRecords` as main (the CAMEL-24191 catch). A new option on a patch branch is your call; the docs and generated files would be backported with it. Tests: - `KclKinesis2ConsumerFailedRecordTest` (new, real route, mocked `RecordProcessorCheckpointer`): with the option, the checkpoint stops before the failed record (also in later batches and at shutdown); a failed first record is not checkpointed; the next lease holder routes the failed record again and checkpoints once it succeeds; a record that always fails keeps the checkpoint before it; the end of the shard is still checkpointed after a failure; the failed record is reported to the exception handler. Without the option, the failed record is reported to the exception handler and the batches are checkpointed as before. With the option ignored (main behaviour) the 5 option tests fail; without the exception handler call the 2 handler tests fail. 20x: 180/180 (both test classes, 20 runs). - `KclKinesis2ConsumerRecordProcessingTest` (CAMEL-24191) passes; only its comments are updated (KCL does not deliver the batch again). - camel-aws2-kinesis module unit tests: 54 run, 0 failures (offline). # 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-26599](https://issues.apache.org/jira/browse/CAMEL-26599). # 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-aws2-kinesis and the modules it depends on, with camel-package-maven-plugin built from this tree, the formatter and import-sort plugins, and ran its unit tests; the endpoint and component DSL were generated with the same plugin's DSL generators from the catalog JSON. 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]
