davsclaus commented on code in PR #27754:
URL: https://github.com/apache/camel/pull/27754#discussion_r4255125530
##########
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:
Question, not blocking: in the `routed == 0` case the iterator of the
request is kept and `lastSeenSequenceNumbers` is not touched. If this shard has
never returned a record before (a quiet table with `FROM_LATEST`, and the first
change arrives while the route is being suspended) and the route stays
suspended for more than 15 minutes, the kept iterator expires and the expired
path calls `requestFreshShardIterator(shardId, null)`, which is `TRIM_HORIZON`,
so the whole retained shard (up to 24h) is routed again. On main this response
would have moved `lastSeenSequenceNumbers` to its last record, so the expired
path would have continued after it. It is narrow (and a quiet shard that
expires already goes to `TRIM_HORIZON` on main), but would it be worth
remembering the first record of the response here, e.g. so the next poll asks
for `AT_SEQUENCE_NUMBER` of `records.get(0)` when there is no last seen
sequence number? Fine as a follow-up if you prefer.
##########
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:
This special case is no longer needed: CAMEL-25559 (#27698, merged
2026-10-12 and already in this branch's base) changed
`PooledExchangeFactory.release` so that an auto-release exchange that was never
routed is offered to the pool once (it takes the on-done task off, calls
`done()`, puts it back and offers it), and an exchange that is already done is
not offered again. So `releaseExchange(notRouted, true)` is now correct for
pooled and non-pooled exchanges, and the comment here about a second offer
describes the old behaviour. Calling `releaseExchange` directly keeps this
consumer on the same path as the other consumers, and
`Ddb2StreamConsumerSuspendedBatchPooledExchangeTest` should still pass as is
(taken == given back). The PR description's sentence about "current main" can
be updated too.
--
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]
