gnodet-bot commented on code in PR #27754:
URL: https://github.com/apache/camel/pull/27754#discussion_r4256152927
##########
components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java:
##########
@@ -67,6 +75,11 @@ protected int poll() throws Exception {
getEndpoint().getConfiguration().getMaxResultsPerRequest()
/ shardIterators.size());
String shardId = shardIteratorEntry.getKey();
String shardIterator = shardIteratorEntry.getValue();
+ if (shardsOfUnfinishedBatch.contains(shardId)) {
+ // not requested when the batch was cut short: that thread may
have been interrupted by the stop
Review Comment:
💡 **Suggestion:** The comment `that thread may have been interrupted by the
stop` is technically not wrong, but it focuses on *why the iterator wasn't
updated* rather than *what this block does*. Consider a clearer framing that
explains the structural reason:
```suggestion
// cached iterator points before already-routed records;
request a fresh one starting from where we left off
```
##########
components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java:
##########
@@ -88,11 +113,33 @@ protected int poll() throws Exception {
for (Record polledRecord : records) {
exchanges.add(createExchange(polledRecord));
}
- processedExchangeCount += processBatch(CastUtils.cast(exchanges));
+ int routed = processBatch(CastUtils.cast(exchanges));
+ processedExchangeCount += routed;
+
+ if (routed < records.size()) {
+ // the route is being suspended or stopped, and processBatch
did not route the rest of the records:
+ // do not move past them, so that they are routed when the
consumer runs again. When nothing was
+ // routed, the shard iterator of this request is kept and used
again.
+ if (routed > 0) {
+ lastSeenSequenceNumbers.put(shardId, records.get(routed -
1).dynamodb().sequenceNumber());
+ firstNotRoutedSequenceNumbers.remove(shardId);
+ shardsOfUnfinishedBatch.add(shardId);
+ } else if (!lastSeenSequenceNumbers.containsKey(shardId)) {
+ // only needed when the kept iterator has expired before
the consumer runs again
+ firstNotRoutedSequenceNumbers.put(shardId,
records.get(0).dynamodb().sequenceNumber());
Review Comment:
💡 **Suggestion:** Consider adding an inline comment explaining why
`firstNotRoutedSequenceNumbers` is only populated when
`lastSeenSequenceNumbers` has no entry for this shard. When
`lastSeenSequenceNumbers` already contains the shard, the
`AFTER_SEQUENCE_NUMBER` iterator from `requestFreshShardIterator` will already
start after the last routed record, so `firstNotRoutedSequenceNumbers` would be
redundant:
```suggestion
} else if (!lastSeenSequenceNumbers.containsKey(shardId)) {
// only record the first-not-routed position when there
is no lastSeen entry;
// if lastSeen exists, requestFreshShardIterator already
starts after it
firstNotRoutedSequenceNumbers.put(shardId,
records.get(0).dynamodb().sequenceNumber());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
