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)
