Federico Mariani created CAMEL-24154:
----------------------------------------
Summary: camel-aws2-ddb - DDB Streams consumer never subscribes to
shards created after startup - silent data loss on resharding
Key: CAMEL-24154
URL: https://issues.apache.org/jira/browse/CAMEL-24154
Project: Camel
Issue Type: Bug
Components: camel-aws
Reporter: Federico Mariani
Attachments: ShardIteratorHandlerReshardingTest.java
Found during a code review of camel-aws2-ddb on main (4.22.0-SNAPSHOT).
In {{ShardIteratorHandler.getShardIterators()}} (ddbstream, lines ~49-77),
{{shardTree.populate(...)}} is only called when
{{currentShardIterators.isEmpty()}} — i.e. on the first poll, or after *all*
tracked shards have been drained and removed. On every later poll the
else-branch resolves successors of a closed shard against that stale snapshot:
{{shardTree.getChildren(shardId)}}. Children created by DynamoDB's routine
shard rotation (roughly every 4 hours) or by a split *after* the initial
{{describeStream}} are not in the tree, so {{children.isEmpty()}} is true and
the closed shard is treated as "still an active leaf". Once its records are
drained, {{updateShardIterator(shardId, null)}} removes it and its lineage is
never followed.
While at least one other shard remains active the iterator map never empties,
so the tree is never refreshed:
* On a table with 2+ partitions, after any partition's shard rotates, *all
subsequent writes to that partition are silently never delivered* — for hours,
until every other shard also happens to close.
* On a single-partition table the map does empty, but re-initialization uses
the *configured* iterator type: with the default {{FROM_LATEST}} the records
written between the parent closing and re-initialization are skipped; with
{{FROM_START}} the entire 24h trim-horizon backlog is re-delivered as
duplicates every ~4h.
Related defects in the same class (kept here for context, listed in the
umbrella): {{getShardIterators()}} returns the internal HashMap by reference
while {{updateShardIterator}} structurally modifies it during the consumer's
iteration (ConcurrentModificationException when a shard retires);
{{describeStream}} pagination ({{lastEvaluatedShardId}}) is ignored so streams
with >100 shards lose partitions; and {{requestFreshShardIterator}} sends
{{AFTER_SEQUENCE_NUMBER}} with a null sequence number when the iterator expired
before any record was seen, leaving the consumer permanently stuck.
*Reproducer*: attached {{ShardIteratorHandlerReshardingTest.java}} (unit test
with a mutable mock streams client, goes under
{{components/camel-aws/camel-aws2-ddb/src/test/java/org/apache/camel/component/aws2/ddbstream/}}).
Two active shards A and B; A rotates into children A1/A2; after A is drained
the handler keeps polling only B:
{noformat}
children of the rotated shard-A must be consumed, but the handler returned
only: {shard-B=iterator-B}
{noformat}
*Suggested fix*: when a tracked shard closes (children lookup empty or
{{nextShardIterator == null}}), re-issue a paginated {{describeStream}} to
refresh the tree and start the closed shard's children at {{TRIM_HORIZON}}.
_This issue was researched and filed by Claude Code (AI) on behalf of Federico
Mariani (fmariani). Full findings document from the review is attached to the
umbrella issues._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
