shashank created CAMEL-25741:
--------------------------------
Summary: camel-aws2-ddb - aws2-ddbstream consumer leaves a closed
shard before reading it to its end, and starts again from streamIteratorType
when every shard it reads has closed (records lost or routed again)
Key: CAMEL-25741
URL: https://issues.apache.org/jira/browse/CAMEL-25741
Project: Camel
Issue Type: Bug
Components: camel-aws
Reporter: shashank
Assignee: shashank
{{ShardIteratorHandler.getShardIterators()}} (camel-aws2-ddb, main
94ca50181230, lines 53-96) decides which shards the {{aws2-ddbstream}} consumer
reads at each poll from two signals, and neither means that a shard was read:
# *A tracked shard that has children in the shard tree is replaced by its
children* (lines 71-80, "Inactive shard, move down to its children"). When
DynamoDB closes a shard it creates its child shards, which {{DescribeStream}}
then lists with the closed shard as {{ParentShardId}}; the closed shard still
holds the records the consumer has not read yet, and the [DynamoDB Streams
guide|https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html]
says "an application must always process a parent shard before it processes a
child shard". {{GetRecords}} returns no {{NextShardIterator}} only when a
closed shard was read to its end
([GetRecords|https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_streams_GetRecords.html]:
"If set to null, the shard has been closed and the requested iterator will not
return any more data"). Consequences:
#* {{streamIteratorType=FROM_START}}: the first poll populates the tree with
every shard of the last 24 hours. The roots are closed and already have
children, so the second poll leaves every root after one response
({{maxResultsPerRequest}} records, default 100, divided by the number of
shards). The rest of the root shards is never routed.
#* any iterator type, table with several partitions: when one shard is read to
its end, the tree is refreshed; every other tracked shard that has closed
meanwhile (a shard has a lifespan of up to four hours, [AWS Database
Blog|https://aws.amazon.com/blogs/database/build-scalable-event-driven-architectures-with-amazon-dynamodb-and-aws-lambda/])
is left with its unread records.
# *An empty map of shard iterators re-initialises the consumer from
{{streamIteratorType}}* (line 58). The map is empty when every shard the
consumer was reading has been read to its end: a table with one partition (one
open shard) at each shard rotation, or several shards read to their end in the
same poll. {{pendingClosedShards}} (CAMEL-24154) is not consulted on that path.
Consequences:
#* {{FROM_START}}: the consumer reads the stream again from its roots and
routes their first records again (then signal 1 moves it down the tree one
response per shard); when the closed roots are read to their end in one
response, the map is empty again at the next poll, so the consumer routes them
again at every poll and never reads the child shards.
#* {{FROM_LATEST}} (the default): the consumer takes {{LATEST}} iterators of
the new shards and skips the records that were written to them between the
rotation and the poll.
Example (one partition, default {{FROM_LATEST}}): the consumer reads
{{shard-0}}; DynamoDB rotates it into {{shard-1}}; the application writes
record 4 (into {{shard-1}}); the consumer reads the last record of {{shard-0}}
(no next iterator), the map is empty, the next poll takes {{LATEST}} of
{{shard-1}}: record 4 is never routed. CAMEL-24154 described this case ("On a
single-partition table the map does empty, but re-initialization uses the
configured iterator type") and its fix covers only the case where another shard
is still tracked.
*Fix* (one class): the configured iterator type is used once, for the start
position (again only while no shard was found at all); a shard is left only
when {{GetRecords}} returned no next shard iterator, and its children are then
read from {{TRIM_HORIZON}} (as before); a shard read to its end whose children
{{DescribeStream}} does not list yet is looked up again at the next poll. The
existing unit test
{{ShardIteratorHandlerTest.shouldProgressThroughTreeWhenShardIteratorsAreRetrievedRepeatedly}}
expected the move to the children at each call without any record read; it now
reads each shard to its end before the next call.
*Tests*: new {{Ddb2StreamConsumerShardRotationTest}} (5 tests, a real route and
an in-memory {{DynamoDbStreamsClient}} with closed and open shards): a closed
root with 5 records and {{maxResultsPerRequest=2}} ({{FROM_START}}; main routes
1, 2 and the child's record only); a closed root and its child ({{FROM_START}};
main routes 1, 2, 1, 2, ... and never 3); the child of the only shard
({{FROM_LATEST}}; main never routes record 4); two shards that rotate together
({{FROM_LATEST}}; main skips five records of the longer shard); a child listed
only after the consumer read the end of its parent ({{FROM_LATEST}}; main skips
its records). All five end with an error on main (an Awaitility
{{ConditionTimeoutException}} around the assertion; the second test routed 2906
records instead of 3 in 20 seconds) and pass with the fix; the camel-aws2-ddb
suite passes (82 tests).
*How it was found*: deterministic simulation testing with a fake DynamoDB
Streams world written from the DynamoDB Streams API reference (shard history,
writes, rotations and splits, faults on every call; the seed also picked
{{streamIteratorType}}, {{maxResultsPerRequest}} 1 to 1000, the poll delay,
{{greedy}} and the {{DescribeStream}} page size). The invariants "every record
the consumer is expected to read is routed" and "no record is routed twice"
failed in 1253 of 3000 seeds on main (first failing seed 3, invariant "every
expected record is routed", {{FROM_START}}); with the fix, 0 of 3000 seeds.
*Affected*: main, camel-4.23.x and camel-4.22.x (same code, both signals);
camel-4.18.x and camel-4.14.x have signal 1 and signal 2 without the
CAMEL-24154 refresh (read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
