[
https://issues.apache.org/jira/browse/CAMEL-25942?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-25942.
---------------------------------
Resolution: Fixed
> camel-aws2-kinesis - resuming with the resumable EIP gives every shard the
> position of another shard, and fails with a ClassCastException once an offset
> was stored
> -------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25942
> URL: https://issues.apache.org/jira/browse/CAMEL-25942
> Project: Camel
> Issue Type: Bug
> Components: camel-aws2-kinesis
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> The polling consumer asks the route's {{ResumeActionAwareAdapter}} to resume
> a shard each time the shard needs a new shard iterator
> ({{Kinesis2Consumer.resume}}, main 9f839f4bcc9d). The adapter calls
> {{evalEntry}} of the {{KinesisResumeAction}} for *every* entry of the resume
> cache, and the default action sets the shard id and sequence number of each
> entry in turn on the {{GetShardIteratorRequest}} of the shard that asked
> ({{KinesisResumeAction.evalEntry}}: {{builder.shardId((String) shardId)}},
> {{AFTER_SEQUENCE_NUMBER}}, {{(String) sequenceNumber}}). So:
> * with more than one shard in the cache, every shard gets the position of the
> last entry. With the cache {{{shard-0=1, shard-1=2}}} and two shards, both
> slots read {{shard-1}} after sequence 2: its records are routed twice (once
> with {{CamelAwsKinesisShardId=shard-0}}) and {{shard-0}} is never read from
> its own position;
> * {{KinesisResumeStrategy.updateLastOffset}} stores a {{KinesisOffset}} in
> the cache when the resumable EIP completes an exchange, but {{evalEntry}}
> casts the value to {{String}}. The next time a shard needs a new iterator
> (the route or context restarts with the same cache, the iterator expired, a
> child shard after resharding), {{getShardIterator}} throws
> {{ClassCastException: KinesisOffset cannot be cast to String}} on every poll
> and the shard is never read again;
> * the adapter is one instance for the route, and {{poll()}} gets the
> iterators of the shards in parallel ({{parallelStream}}). {{setResumeAction}}
> + {{resume()}} of two shards can interleave, and one shard is then resumed
> with the action (and request builder) of the other.
> CAMEL-24156 listed the multi-shard resume in its camel-aws2-kinesis section;
> PR #24863, which resolved that umbrella, only created one action per shard.
> The deleted {{KinesisDefaultResumeAdapter}} (before CAMEL-20840) looked up
> {{cache.get(shardId, KinesisOffset.class)}}; the action that replaced it
> ignores its own {{shardId}}. The only test, {{KinesisConsumerResumeIT}},
> overrides {{evalEntry}} and uses one shard.
> *Fix*: the default action uses only the entry whose key is its own shard id,
> takes the sequence number from a {{String}} or an {{Offset}}
> ({{KinesisOffset}}), and keeps the entry ({{false}}, per the {{ResumeAction}}
> contract). The consumer sets the action and resumes under one lock shared by
> all consumers, because routes that use the same resume strategy bean share
> its adapter and {{poll()}} gets the iterators of the shards in parallel. One
> paragraph in the component docs (Custom Resume Action) says that
> {{evalEntry}} sees every entry and that a subclass should compare the key
> with {{getShardId()}}.
> *Tests*: three new tests in {{KinesisConsumerPositionTest}} (in-memory
> Kinesis client of that test; a resume cache that follows the
> {{ResumeCache.forEach}} contract like camel-caffeine's {{CaffeineCache}}).
> All fail on main:
> * {{resumeStrategyStartsEachShardAfterItsOwnSequenceNumber}}: {{expected:
> <[b0, c0, c1]> but was: <[c1, c1]>}};
> * {{resumeStrategyStartsANewConsumerAfterTheOffsetsOfTheResumableEip}}
> (offsets stored as the resumable EIP does, new consumer, then an expired
> iterator): {{ClassCastException: KinesisOffset cannot be cast to String}} at
> {{KinesisResumeAction.evalEntry}};
> * {{consumersOfOneResumeStrategyResumeOneAfterTheOther}} (two consumers with
> one strategy): {{the second consumer set its resume action while the first
> one resumed with its own}}.
> camel-aws2-kinesis unit tests with the fix: 49 run, 0 failures.
> Not changed (follow-up): {{TransientResumeStrategy.createSimpleCache()}}
> (used by the IT) removes an entry when the action returns {{false}}, the
> opposite of the {{ResumeAction}} contract that {{CaffeineCache}} follows, and
> removes it from the {{HashMap}} while iterating it, so with two or more
> entries the resume throws {{ConcurrentModificationException}}.
> *How it was found*: while checking the open items of CAMEL-24156 on main, by
> reading the resume path of the polling consumer. The JUnit tests above are
> the evidence.
> *Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x, camel-4.14.x,
> camel-4.10.x and camel-4.8.x (the same {{KinesisResumeAction.evalEntry}};
> read, not run).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
