[
https://issues.apache.org/jira/browse/CAMEL-25569?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25569:
--------------------------------
Fix Version/s: 4.24.0
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> {{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)
