Croway commented on PR #29272: URL: https://github.com/apache/camel/pull/29272#issuecomment-6669783295
@davsclaus thanks, I checked both points against the PR head. **1. Graceful shutdown.** Confirmed, the chain is as you describe: - `Kinesis2Consumer.processBatch` (line 245) never calls `isBatchAllowed()`. - `ScheduledBatchPollingConsumer.getPendingExchangesSize()` returns 1 while `isPolling()` is true. - Suspending the consumer does not stop the poll that is running (`ScheduledPollConsumer.doSuspend()` is empty). - `DefaultShutdownStrategy` times out after 45 s by default and then forces the routes to shut down. A local run with a mock `KinesisClient`: shutdown timeout 5 s, a route that takes 20 ms per record, and `stopRoute` during the first poll. | | `stopRoute` took | forced | records routed | records failed after the stop | |---|---|---|---|---| | default (1000), 1 shard | 5.0 s | yes | 209 of 1000 | 791 (790 `RouteStoppingException`, 1 interrupted) | | default (1000), 4 shards | 5.0 s | yes | 823 of 4000 | 3177 | | `maxResultsPerRequest=1` | 0 s | no | 1 of 1 | 0 | Done in 6b9d60fd4f0b: - The upgrade-guide entry has a new paragraph with your 50 ms / 45 s example. It says that the shutdown is then forced, and that the records of the poll that are not routed yet fail with `RouteStoppingException`. - The throughput section of the component doc has a matching paragraph. The catalog copy is regenerated. The `isBatchAllowed()` change is not in this PR. It is drafted as a follow-up: CAMEL-26669. It is the same change that CAMEL-25570 made for aws2-ddbstream. From the evaluation: - Both parts must go in together. The check alone would make a `suspendRoute` / `resumeRoute` skip the rest of the response, because the same consumer continues from the stored iterator. That is the bug CAMEL-25570 fixed. - When no record of the response was routed, the follow-up should request `AT_SEQUENCE_NUMBER` for the first record instead of using the cached iterator again: the Kinesis API docs do not say that an iterator can be used twice. - A null `nextShardIterator` must not be stored for a response that was cut short. Otherwise a closed shard is marked closed before its last records are routed. - The size is about 30-50 lines in `Kinesis2Consumer`, plus tests. - A route that is stopped and started gets a new consumer. For a stop, the gain is a prompt stop; the kept position matters for suspend and resume. **2. The common pool.** Confirmed. With 4 shards on a 10-CPU machine, 620 of the 827 records that entered the route ran on `ForkJoinPool.commonPool-worker-N`, and the others on the scheduler thread. With one shard, everything runs on the scheduler thread. In the forced stop above, only the scheduler thread was interrupted: the common-pool workers went on and failed their remaining records. This is drafted as CAMEL-26670: read the shards on an executor from the `ExecutorServiceManager`, sized by an option. The default is still to be decided. Today a poll reads up to the common-pool parallelism + 1 shards at a time, and that parallelism is `availableProcessors() - 1`, at least 1. _Claude Code on behalf of Federico Mariani_ -- 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]
