[jira] [Created] (CAMEL-25569) camel-aws2-kinesis - after an expired shard iterator the consumer starts the shard over (TRIM_HORIZON: all records again, LATEST: records skipped), and with LATEST it skips the first records of a shard created by resharding

["shashank (Jira)" <ji...@apache.org>]

shashank created CAMEL-25569:
--------------------------------

             Summary: camel-aws2-kinesis - after an expired shard iterator the consumer starts the shard over (TRIM_HORIZON: all records again, LATEST: records skipped), and with LATEST it skips the first records of a shard created by resharding
                 Key: CAMEL-25569
                 URL: https://issues.apache.org/jira/browse/CAMEL-25569
             Project: Camel
          Issue Type: Bug
          Components: camel-aws2-kinesis
            Reporter: shashank
            Assignee: shashank


{{Kinesis2Consumer.getShardIterator}} requests every new shard iterator with the configured {{iteratorType}} (line 265 at main 5987a263107b). That is the position to start a shard at when the consumer starts. It is also used in two cases where the consumer already has a position:

*1. The cached iterator has expired.* A shard iterator is valid for 5 minutes. Since CAMEL-24156 the consumer catches {{ExpiredIteratorException}}, removes the cached iterator and calls {{getShardIterator}} again (lines 159-175), which requests the configured type:
* {{TRIM_HORIZON}} (the default): the consumer routes all the records of the shard again, from the oldest record of the retention period (24 hours by default).
* {{LATEST}}: the records stored since the last poll are skipped.
* {{AT_TIMESTAMP}}, {{AT_SEQUENCE_NUMBER}}, {{AFTER_SEQUENCE_NUMBER}}: the consumer goes back to the configured position.
The only log line is the WARN "Shard iterator expired ..., requesting a fresh one". An iterator expires when the consumer does not poll the shard for 5 minutes: the route was suspended for longer ({{suspendRoute}} / {{resumeRoute}}, the same consumer continues), {{delay}} or the back-off ({{backoffMultiplier}}) is above 5 minutes, or the route takes that long for the records of one poll. (A route that is stopped and started gets a new consumer, which starts from the configured type; that is not changed.) With a resume strategy the request is then changed by its adapter, as for the first iterator.

*2. A shard created by resharding.* When a shard is split or merged while the consumer runs, the shard monitor (every 10 seconds by default, lines 466-477) adds the new shards to the list, and the consumer requests their first iterator with the configured type. With {{LATEST}} the records stored in a new shard before that request are skipped. The consumer was reading the parent shard, so every record of the new shard is new for it.

For comparison, the aws2-ddbstream consumer of the same component family does both on main: after an expired iterator it requests {{AFTER_SEQUENCE_NUMBER}} with the last sequence number it saw ({{Ddb2StreamConsumer}} lines 77-85, {{ShardIteratorHandler.requestFreshShardIterator}} lines 108-115), and it reads child shards from {{TRIM_HORIZON}} ({{ShardIteratorHandler}} line 77).

h3. Reproduction

New {{KinesisConsumerPositionTest}} (5 tests): the Kinesis client is a Mockito mock backed by a small in-memory stream (shards with records, iterators that are positions and can be expired, parent/child shards); the scheduler is not started and the test calls {{poll()}}. On main 3 tests fail:
* {{expiredIteratorContinuesAfterTheLastRoutedRecord}} ({{TRIM_HORIZON}}; a, b routed; iterators expire; c stored): {{only the record that was not routed yet ==> expected: <1> but was: <3>}} (a and b are routed again).
* {{expiredIteratorDoesNotSkipTheRecordsOfTheGapWithLatest}}: {{the record stored while the iterator expired ==> expected: <1> but was: <0>}}.
* {{shardCreatedByReshardingIsReadFromItsStartWithLatest}}: {{the records stored in the child shards before the consumer saw these shards ==> expected: <[a, b, c]> but was: <[a]>}} (the records of the two child shards are skipped).
The other two pass on main and guard what must not change: a shard without routed records gets the configured type after an expiry, and the shards of the first shard list are read from {{LATEST}} also when they are child shards of an earlier resharding.

h3. Proposed fix

In {{Kinesis2Consumer}}:
* remember the sequence number of the last record of each response per shard, and request {{AFTER_SEQUENCE_NUMBER}} with it when a new iterator is needed for that shard;
* in the shard monitor, mark a shard that appears after the first list and descends from a known shard, and request {{TRIM_HORIZON}} for it;
* otherwise the configured type, as before. The resume strategy is applied after that, as before.

Known limit: with {{LATEST}}, when the iterator of a shard expires before any record was routed from it, the new iterator is {{LATEST}} again and the records stored in between are skipped, as on main. The three per-shard collections of the fix ({{lastSequenceNumbers}}, {{knownShardIds}}, {{childShardIds}}) are not pruned when a shard leaves the shard list, like {{currentShardIterators}} on main.

Behaviour change, with a 4.24 upgrade-guide entry: no repeated or skipped records after an expired iterator; child shards are read from their start whatever the {{iteratorType}}. No new option. camel-aws2-kinesis unit tests with the fix: 46, 0 failures (the integration tests need LocalStack and were not run).

Found with a TLA+ model of the consumer position (shards as record logs, iterator expiry between polls, a split, the shard monitor): on main TLC violates {{NoDuplicate}} with {{TRIM_HORIZON}} (13 states: route record 1, the iterator expires, the next poll gets a new iterator at the start, record 1 is routed again) and {{NoSkip}} with {{LATEST}} (after an expiry, and for the child shard after a split: 10 states); both hold with the fix, except the known limit above. Then confirmed with the real consumer as described.

Affected: main and camel-4.22.x (the expired-iterator handling of CAMEL-24156; read, not run). camel-4.18.x and camel-4.14.x do not catch {{ExpiredIteratorException}} (the consumer fails every poll instead) and have the resharding case (line 260 on camel-4.18.x; read, not run).

Duplicate check (2026-10-12): JIRA text "kinesis shard iterator expired" (1: CAMEL-24156, which added the handling this ticket is about), "Kinesis2Consumer AFTER_SEQUENCE_NUMBER" (1: CAMEL-24156), "kinesis child shard" (none), "kinesis resharding" (3: CAMEL-19789 repeated first record, CAMEL-12179 KCL consumer, CAMEL-24156; none about the position of new shards), "kinesis split shard" (4, same tickets plus CAMEL-24197 and CAMEL-24191); GitHub pull requests "Kinesis2Consumer expired iterator": #24863 (merged, CAMEL-24156); no open pull request touches {{Kinesis2Consumer}} (#26871 touches {{KclKinesis2Consumer}} only).

_Filed with Claude Code on behalf of allthingssecurity._




--
This message was sent by Atlassian Jira
(v8.20.10#820010)