allthingssecurity commented on code in PR #27754:
URL: https://github.com/apache/camel/pull/27754#discussion_r4256023113
##########
components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java:
##########
@@ -125,6 +153,16 @@ public int processBatch(Queue<Object> exchanges) throws
Exception {
return answer;
}
+ private void releaseNotRoutedExchange(Exchange exchange) {
+ if (exchange instanceof PooledExchange pooled) {
+ // created with autoRelease, so done() also gives it back to the
pool, once: releaseExchange would then
+ // add it to the pool a second time, and two later records would
share the exchange
+ pooled.done();
+ } else {
+ releaseExchange(exchange, true);
+ }
+ }
Review Comment:
Done in 54e1ca5f320c. I checked first that 28e150782cdc (CAMEL-25559,
#27698) is an ancestor of the branch and that
`PooledExchangeFactory.release` there offers a never-routed auto-release
exchange once. `releaseNotRoutedExchange` is
removed and the loop calls `releaseExchange(notRouted, true)`.
`Ddb2StreamConsumerSuspendedBatchPooledExchangeTest`
passes unchanged against the core modules built from the branch (4 taken and
4 given back, then 6 and 6, none
discarded); camel-aws2-ddb: 77 tests, 0 failures. The sentence about
"current main" in the description is replaced.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java:
##########
@@ -88,7 +99,24 @@ 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());
+ shardsOfUnfinishedBatch.add(shardId);
+ }
Review Comment:
Good catch, done in ce96a9f0c922. When nothing was routed and no record of
the shard was routed before, the consumer
remembers the sequence number of the first record of the response. If the
kept iterator has expired at the next poll,
the expired path requests `AT_SEQUENCE_NUMBER` for that record instead of
`TRIM_HORIZON`. With a last seen sequence
number the path is as before (`AFTER_SEQUENCE_NUMBER`), and a kept iterator
that is still valid is used again without
a new request.
New test
`firstRecordsOfAShardThatWereNotRoutedAreNotReplacedByTheWholeShardWhenTheIteratorExpires`:
a `FROM_LATEST`
consumer, two older records in the shard, the first response with records
arrives while the route is suspended, the
iterators expire, the route is resumed. Without the change: `expected: <[3,
4]> but was: <[1, 2, 3, 4]>`.
The branch also has a merge of main (c7b31ccf055d) for the upgrade guide
conflict; both sides are kept and the rest of
the diff is unchanged.
_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]
