allthingssecurity opened a new pull request, #27754: URL: https://github.com/apache/camel/pull/27754
# Description [CAMEL-25570](https://issues.apache.org/jira/browse/CAMEL-25570) `Ddb2StreamConsumer.poll` gets up to `maxResultsPerRequest` records (100 by default) of a shard, creates one exchange per record and calls `processBatch`. After that it stores the next shard iterator of the response, and the sequence number of the last record of the response, whatever `processBatch` has routed. `processBatch` stops when `isBatchAllowed()` returns false. `ScheduledBatchPollingConsumer.isBatchAllowed` (lines 104-118) does that when the consumer is no longer allowed to run (stopping or stopped), or when `deferShutdown` gave it a `shutdownRunningTask` other than `CompleteAllTasks`. Only `DefaultShutdownStrategy` calls `deferShutdown` (lines 614-616), when it suspends or shuts down routes, with the `shutdownRunningTask` of the route (`CompleteCurrentTaskOnly` by default). So the batch is cut, after the exchange that is being routed, by a suspend at route level: - `RouteController.suspendRoute` (`AbstractCamelContext` line 1651), also when a route policy calls it (`RoutePolicySupport.suspendRoute`, line 156); - `CamelContext.suspend()` (`AbstractCamelContext` line 2325). It is not cut by `consumer.suspend()` called directly: `isRunAllowed()` stays true for a suspended service (`BaseService` lines 349-351) and no `shutdownRunningTask` is set, so the poll that is running routes all its records. That is what `RoutePolicySupport.suspendOrStopConsumer` does (line 110), which the route policies shipped with Camel use (`ThrottlingInflightRoutePolicy`, `ThrottlingExceptionRoutePolicy`, the quartz `ScheduledRoutePolicy`), so they are not affected. A route with `shutdownRunningTask(CompleteAllTasks)` is not affected either. When the route is resumed (`resumeRoute`, or `CamelContext.resume()` after `suspend()`), the same consumer continues with the stored iterator, which is after the whole response: the records that were not routed are skipped, and nothing is logged. With one shard and the defaults that is up to 99 records per suspend. The following shards of that poll are requested as well and lose their whole response. When the route or the context is stopped, the batch is cut in the same way (the same `deferShutdown` call), but that consumer is not used again: a route that is started again gets a new consumer (`DefaultRoute.gatherRootServices`, line 711), which starts from the configured `streamIteratorType`. Nothing is skipped there because of the stored iterator. This change: `poll` compares the number of exchanges `processBatch` routed with the number of records of the response. When the batch was cut short, - it does not take the next shard iterator of the response; - when some records were routed, it stores the sequence number of the last routed record, and the next poll requests an iterator after it (`ShardIteratorHandler.requestFreshShardIterator`, which the expired-iterator path already uses). The request is made by the next poll and not at once, because a stop interrupts the polling thread; - when nothing was routed, the shard iterator of the request stays and is used again (if it expires meanwhile, the existing expired-iterator path continues after the last record of the previous response); - the following shards are not requested by that poll. The exchanges created for the records that are not routed are now given back to the exchange factory (before, they were dropped). `poll` reads nothing from them: the sequence number comes from the record. For a pooled exchange it calls `done()` and not `releaseExchange(exchange, true)`: these exchanges are created with `createExchange(true)`, so `done()` already gives them back to the pool, and `releaseExchange` on an exchange that was never routed would offer it a second time on current main (that is what #27698 changes). `Ddb2StreamConsumerSuspendedBatchPooledExchangeTest` runs the same tests with `PooledExchangeFactory` and checks that every exchange taken is given back once. There is an entry in the 4.24 upgrade guide (`camel-4x-upgrade-guide-4_24.adoc`; if that file is not on main yet, this PR creates it with the header of the 4_23 guide and adds it to the list in `camel-4x-upgrade-guide.adoc`). Other consumers have the same shape (a cursor moved for the whole poll, then a `processBatch` loop on `isBatchAllowed()`): `SlackConsumer` (the timestamp is set in `createExchanges`) and `GoogleCalendarStreamConsumer` (`selectUndeliveredAndMoveCursor`). I read them on main but did not run them; they are in other modules and not part of this PR. `Kinesis2Consumer.processBatch` has no `isBatchAllowed()` check. Tests: - New `Ddb2StreamConsumerSuspendedBatchTest` (3 tests) with a real route and an in-memory `DynamoDbStreamsClient` (one shard, iterators are positions). The processor of one record, or the `GetRecords` call, suspends the route from another thread (`RouteController.suspendRoute`) and waits until `isBatchAllowed()` is false; then the route is resumed. There is no sleep in the test: it waits on conditions. - Without the change all 3 end with an error, not a failure (`Tests run: 3, Failures: 0, Errors: 3`): the assertion is inside an Awaitility `untilAsserted`, so surefire reports the `ConditionTimeoutException` that wraps it, for example `ConditionTimeout Assertion condition defined as a Lambda expression in org.apache.camel.component.aws2.ddbstream.Ddb2StreamConsumerSuspendedBatchTest records routed after the resume ==> expected: <[1, 2, 3, 4]> but was: <[1, 2]> within 20 seconds.` The other two end the same way with `expected: <[1, 2]> but was: <[]>` (suspend while the request is in flight) and `expected: <[1, 2, 3, 4]> but was: <[1, 4]>`. - New `Ddb2StreamConsumerSuspendedBatchPooledExchangeTest` (the 3 tests again with `PooledExchangeFactory`, and 1 test on the pool statistics: 4 exchanges taken and 4 given back when the batch was cut after 2 of 4 records, 6 and 6 after the resume, none discarded). - With the change, the camel-aws2-ddb unit tests pass: 75 tests, 0 failures. The new test classes passed three times in a row. The integration tests (LocalStack) were not run. Known limits, not changed: - `ShardIteratorHandler.getShardIterators` replaces a shard by its child shards as soon as the shard tree shows children for it (main, lines 71-80). If that happens to the shard whose batch was cut short, the rest of that shard is not read (read in the code, not run), as on main for any shard that still has unread records at that moment, and its id stays in the new `shardsOfUnfinishedBatch` set (one string; `lastSeenSequenceNumbers` keeps an entry per closed shard in the same way). Found with a TLA+ model of a poll (GetRecords, route one record at a time while the batch is allowed, end of the batch) with a suspend and a resume of the route as environment steps. On main TLC violates "between two polls the consumer has not moved past a record it did not route" in 4 states: GetRecords returns 3 records, the route is suspended, the batch ends, the iterator is after record 3. The property holds in the model of the change, also when the iterator expires while the route is suspended. I then reproduced it with the real consumer and a real route. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested the affected module, including the formatter and import-sort plugins. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
