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