allthingssecurity commented on code in PR #27754:
URL: https://github.com/apache/camel/pull/27754#discussion_r4256635015
##########
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:
Thanks. I would like to keep this comment as it is: what the block does is
readable from the two lines below it, while the comment gives the reason the
fresh iterator is requested here and not at the point where the batch was cut
short (that thread may have been interrupted by the stop, so a request made
there can fail). The suggested wording would drop that reason.
_Claude Code on behalf of allthingssecurity_
##########
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:
Thanks. The reasoning is right: when the shard has a last seen sequence
number, the expired-iterator path already continues after it, so the first
not-routed record is only remembered when there is none. The existing comment
on that branch says when the entry is used ("only needed when the kept iterator
has expired before the consumer runs again"), and the condition itself states
the other half. As the PR is approved I am leaving the code untouched; I am
happy to add the extra sentence if a maintainer prefers it.
_Claude Code on behalf of allthingssecurity_
--
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]
