allthingssecurity opened a new pull request, #27753: URL: https://github.com/apache/camel/pull/27753
# Description [CAMEL-25569](https://issues.apache.org/jira/browse/CAMEL-25569) `Kinesis2Consumer.getShardIterator` requests every new shard iterator with the configured `iteratorType`. That is the position to start a shard at when the consumer starts, but it is also used in two cases where the consumer already has a position in the stream: 1. **The cached iterator has expired** (a shard iterator is valid for 5 minutes). Since CAMEL-24156 the consumer catches `ExpiredIteratorException`, drops the cached iterator and requests a new one, of the configured type. With `TRIM_HORIZON` (the default) it then routes all the records of the shard again; with `LATEST` it skips the records stored since the last poll; with a timestamp or a sequence number it goes back to the configured position. This happens when the shard is not polled for 5 minutes: a route that was suspended for longer (`suspendRoute` / `resumeRoute`, the same consumer continues), a `delay` or back-off above 5 minutes, or a route that takes that long for the records of one poll. A route that is stopped and started gets a new consumer and starts from the configured type; that is not changed. 2. **A shard created by a split or a merge while the consumer runs.** The shard monitor adds it to the shard list (every 10 seconds by default) and its first iterator is requested with the configured type. With `LATEST` the records stored in the new shard before that are skipped, although the consumer was reading the parent shard. The aws2-ddbstream consumer already does both on main: after an expired iterator it requests `AFTER_SEQUENCE_NUMBER` with the last sequence number it saw (`Ddb2StreamConsumer.poll`, the `ExpiredIteratorException` catch, and `ShardIteratorHandler.requestFreshShardIterator`), and it reads child shards from `TRIM_HORIZON` (`ShardIteratorHandler.getShardIterators`). This change, in `Kinesis2Consumer`: - it remembers the sequence number of the last record of each response per shard, and requests `AFTER_SEQUENCE_NUMBER` with it when it needs a new iterator for that shard; - the shard monitor marks a shard that appears after the first shard list and descends from a known shard (`parentShardId` / `adjacentParentShardId`), and the consumer requests `TRIM_HORIZON` for it; - in every other case the configured type is used as before, and a resume strategy is applied after that as before. The three new collections (`lastSequenceNumbers`, `knownShardIds`, `childShardIds`) are not cleared in `doStop`, like the cached iterators next to them. Per record nothing is added; per response one map put. Known limit, not changed: with `LATEST`, when the iterator of a shard expires before any record was routed from it, the new iterator is `LATEST` again, so records stored in between are skipped as on main. The fixed `shardId` mode only reads the configured shard, as before. Known limit, not changed: the three collections keep an entry per shard and are never pruned when a shard leaves the shard list (a closed shard after its retention period), like `currentShardIterators` on main. That is a few short strings per shard the stream had while the consumer runs. Behaviour change, with an entry in the 4.24 upgrade guide (`camel-4x-upgrade-guide-4_24.adoc`; if that file is not on main yet, this PR creates it with the header of the 4_23 guide and adds it to the list in `camel-4x-upgrade-guide.adoc`): no repeated or skipped records after an expired iterator, and a shard created by resharding is read from its first record whatever the `iteratorType`. Tests: - New `KinesisConsumerPositionTest` (5 tests). The Kinesis client is a Mockito mock backed by a small in-memory stream (records per shard, iterators that are positions and can expire, parent and child shards). The scheduler is not started, the test calls `poll()`. - Without the change 3 of them fail: `only the record that was not routed yet ==> expected: <1> but was: <3>` (TRIM_HORIZON after an expiry), `the record stored while the iterator expired ==> expected: <1> but was: <0>` (LATEST after an expiry), `the records stored in the child shards before the consumer saw these shards ==> expected: <[a, b, c]> but was: <[a]>` (child shards with LATEST). - The other 2 pass without the change as well. They pin what must stay: a shard without routed records gets the configured type after an expiry, and the shards of the first list are read from `LATEST` also when they are child shards of an earlier resharding. - With the change, the camel-aws2-kinesis unit tests pass: 46 tests, 0 failures. The integration tests (LocalStack) were not run. Found with a TLA+ model of the consumer position (shards as record logs, an iterator that expires between two polls, a split, the shard monitor). On main TLC violates "a record is routed at most once" with `TRIM_HORIZON` (route record 1, the iterator expires, the next poll gets a new iterator at the start of the shard, record 1 is routed again) and "the consumer does not move past a record without routing it" with `LATEST` (after an expiry, and for the child shard after a split). Both hold in the model of the change, except for the known limit above. I then reproduced it with the real consumer. # 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 the affected module, 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]
