Croway opened a new pull request, #29272: URL: https://github.com/apache/camel/pull/29272
JIRA: [CAMEL-26650](https://issues.apache.org/jira/browse/CAMEL-26650) ## What was wrong The polling `aws2-kinesis` consumer (the default, `useKclConsumers=false`) sends `maxResultsPerRequest` as the `Limit` of the `GetRecords` request it sends to each shard in each poll. The option defaulted to `1`. With the default `delay` of 500 ms, a shard was read at about 2 records per second, so any shard that gets more than that falls further behind with every poll. `greedy=true` does not help much: it still gets one record per call, and Kinesis allows 5 `GetRecords` calls per second per shard. ## What changed - `Kinesis2Configuration`: `maxResultsPerRequest` now defaults to `1000`. The description now says the limit applies to each shard in each poll, that it is the `Limit` of `GetRecords` (1 to 10000), and that the KCL consumer does not use it. 1000 was picked over the API maximum of 10,000 because of the memory a poll holds: one response per shard while its records are routed, and up to 10 MiB per response at 10,000. - Component doc: a new section, "Consumer throughput and the Kinesis read quotas", explains how `maxResultsPerRequest`, `delay` and `greedy` relate to the per-shard read quotas. The quotas were checked in the Kinesis service model of AWS SDK 2.57.7 (the version Camel uses): 5 `GetRecords` per second per shard, 2 MiB per second per shard, at most 10,000 records or 10 MiB per response, and a 5-second `ProvisionedThroughputExceededException` window after a 10 MiB response. - The KCL consumer (`KclKinesis2Consumer`) does not read `maxResultsPerRequest`. It extends `DefaultConsumer` and builds its own `PollingConfig`, so it does not use `delay` or `greedy` either. It is not changed. - Regenerated: the component JSON, the catalog JSON and doc copy, and the endpoint-dsl and component-dsl builders. ## Tests - New `KinesisConsumerGetRecordsLimitTest` uses a Mockito `KinesisClient` with one shard holding 1500 records. `GetRecords` returns as many records as its `Limit` asks for. - With the default, two polls route 1000 and then 500 records, and both requests send `Limit=1000`. - With `maxResultsPerRequest=1`, each poll routes one record. - On main, the default case fails with `expected: <1000> but was: <1>`. With this change it passes. - `KinesisEndpointTest.onlyRequiredEndpointParams` asserted the old default of 1. It now expects 1000. - `KinesisConsumerResumeIT` passed on main only because one record came per poll. Its resume cache was empty, so the consumer never called `TestResumeAction`. On main it got records 0-9 instead of 10-19, and it checked only the count. With 1000 records per poll, all 20 records came in one poll and the count check failed (11 instead of 10). The test now adds an entry for the shard to the resume cache, and it checks that every record it gets comes after the resume position. With this change it gets records 10-19. - Module: 59 unit tests pass. These ITs pass on Floci: `KinesisConsumerIT`, `KinesisConsumerResumeIT`, `KinesisProducerIT` and the three `Kinesis2ConsumerHealth*IT`. `KclKinesisConsumerIT` and `KinesisFirehoseComponentManualIT` are `@Disabled` (manual). ## Docs and upgrade guide - `aws2-kinesis-component.adoc`: the new throughput section. - `camel-4x-upgrade-guide-4_24.adoc`: a new entry, "camel-aws2-kinesis - the consumer fetches up to 1000 records per shard in each poll". It says that `maxResultsPerRequest=1` keeps the previous behaviour, and that a poll now holds more records in memory. - The `aws-kinesis-source` Kamelet does not set `maxResultsPerRequest`, so Kamelet users get the new default too. ## Not in this PR The consumer reads the shards with `parallelStream()`, so the `GetRecords` calls and the routing of the records run on the JVM-wide `ForkJoinPool.commonPool()`. The JIRA mentions this too. It is left for a follow-up. ## Backport Not a backport candidate for camel-4.18.x or camel-4.22.x, because it changes a default. Users of those versions can set `maxResultsPerRequest` themselves. _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]
