shashank created CAMEL-25942:
--------------------------------
Summary: 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
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)
