allthingssecurity opened a new pull request, #28461: URL: https://github.com/apache/camel/pull/28461
# Description [CAMEL-25942](https://issues.apache.org/jira/browse/CAMEL-25942) When a shard needs a new shard iterator, the polling consumer asks the route's `ResumeActionAwareAdapter` to resume it. The adapter calls `KinesisResumeAction.evalEntry` for every entry of the resume cache, and the default action set the shard id and sequence number of each entry in turn on the `GetShardIteratorRequest` of the shard that asked. So: - with more than one shard in the cache, every shard got the position of the last entry. With `{shard-0=1, shard-1=2}` both slots read `shard-1` after sequence 2: its records were routed twice (once with the shard id header `shard-0`), and `shard-0` was never read from its own position; - `KinesisResumeStrategy.updateLastOffset` stores a `KinesisOffset` when the resumable EIP completes an exchange, but `evalEntry` cast the value to `String`. The next new iterator of a shard (route or context restarted with the same cache, expired iterator, child shard) failed with `ClassCastException` on every poll, and the shard was never read again; - the adapter is one instance per resume strategy (shared by the routes that use the same strategy bean), and `poll()` gets the iterators of the shards in parallel, so two shards could interleave `setResumeAction` and `resume()`, and one shard was resumed with the action of the other. CAMEL-24156 listed the multi-shard resume; #24863 only created one action per shard. The adapter removed by CAMEL-20840 looked the shard up with `cache.get(shardId, KinesisOffset.class)`. Changes: - `KinesisResumeAction.evalEntry` uses only the entry of its own shard (`getShardId()`), takes the sequence number from a `String` or an `Offset`, and returns `false` (keep the entry, as the `ResumeAction` contract says). - `Kinesis2Consumer.resume` sets the action and resumes under one static `ReentrantLock` for all consumers. A lock per consumer would not do, because two routes with the same strategy bean share its adapter. A `synchronized (adapter)` block would also work, but CAMEL-20199 removed those from camel-aws for virtual threads. The section is held only while a shard gets a new iterator (start, expired iterator, child shard), and with the default adapter it is in memory (`cache.forEach` + `evalEntry`), so one lock for all consumers costs nothing measurable. - Component docs (Custom Resume Action): `evalEntry` sees every entry of the cache, and a subclass should compare the key with `getShardId()`. The catalog copy of the page is updated. - No option, generated file or upgrade guide change. - Backport: the per-shard action of #24863 is on main, camel-4.23.x and camel-4.22.x; camel-4.18.x and older (to 4.8.x) have the same `evalEntry`, but share one registry action between the shards, so a cherry-pick there needs that part too. Not changed (follow-up JIRA): `TransientResumeStrategy.createSimpleCache()` (camel-core-processor, used by `KinesisConsumerResumeIT`) removes an entry when the action returns `false`, the opposite of the `ResumeAction` contract that camel-caffeine's `CaffeineCache` follows, and it removes from its `HashMap` while iterating, so with two or more entries the resume throws `ConcurrentModificationException`. Changing it also changes what `KafkaResumeAdapter` sees with that cache, so I left it for its own JIRA. Tests: - Three new tests in `KinesisConsumerPositionTest`, with that test's in-memory Kinesis client and a resume cache that follows the `ResumeCache.forEach` contract like `CaffeineCache` (comment on the class). All three fail on main: - `resumeStrategyStartsEachShardAfterItsOwnSequenceNumber`: `expected: <[b0, c0, c1]> but was: <[c1, c1]>`; - `resumeStrategyStartsANewConsumerAfterTheOffsetsOfTheResumableEip` (offsets stored as the resumable EIP does through `KinesisConsumerOffsetProcessor`, a new consumer on the same cache, then an expired iterator): `ClassCastException: KinesisOffset cannot be cast to String` in `KinesisResumeAction.evalEntry`; - `consumersOfOneResumeStrategyResumeOneAfterTheOther` (two consumers with one strategy; its adapter keeps the first resume open until another action is set, at most 1 s): `the second consumer set its resume action while the first one resumed with its own ==> expected: <1> but was: <0>`. - camel-aws2-kinesis unit tests: 49 run, 0 failures. Formatter and import sort ran. Built offline with `-Daws-java-sdk2-version=2.55.15` and `-Dasm-version=9.11` because the versions of main were not in my local repository. # 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-25942](https://issues.apache.org/jira/browse/CAMEL-25942). # 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 ran its tests, including the formatter and import-sort plugins. The only generated file is the catalog copy of the component page, updated by hand. 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]
