[
https://issues.apache.org/jira/browse/CAMEL-26650?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Federico Mariani reassigned CAMEL-26650:
----------------------------------------
Assignee: Federico Mariani
> camel-aws2-kinesis - consumer fetches one record per GetRecords call by
> default (maxResultsPerRequest=1)
> --------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26650
> URL: https://issues.apache.org/jira/browse/CAMEL-26650
> Project: Camel
> Issue Type: Improvement
> Components: camel-aws2-kinesis
> Affects Versions: 4.18.5, 4.22.2, 4.23.0
> Reporter: Federico Mariani
> Assignee: Federico Mariani
> Priority: Major
>
> h3. Problem
> The polling {{aws2-kinesis}} consumer is the default
> ({{useKclConsumers=false}}). Each poll it sends one {{GetRecords}} per shard,
> with {{Limit}} = {{maxResultsPerRequest}}, and that option defaults to 1.
> Line numbers are on main (7978db3d42dd):
> *
> {{components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java}}
> lines 52-54: {{maxResultsPerRequest = 1}}.
> * {{Kinesis2Consumer.java}} lines 207-214:
> {{GetRecordsRequest.builder()...limit(getMaxResultsPerRequest())}}.
> So with the defaults ({{delay=500}}, {{greedy=false}}) each shard is read at
> no more than about 2 records per second. The AWS SDK 2.57.7 service model
> sets the limits on the other side:
> * a shard accepts up to 1,000 records per second on write;
> * {{GetRecords}} returns up to 10,000 records (10 MiB) per call, which is
> also the default when no Limit is sent;
> * each shard allows 5 calls per second and 2 MiB per second of reads.
> A shard that receives more than about 2 records per second therefore falls
> further behind with every poll. The consumer never reads
> {{MillisBehindLatest}}, so nothing reports the lag. Records older than the
> stream retention (at least 24 hours, configurable up to 365 days) are trimmed
> before the consumer reaches them. {{greedy=true}} polls again at once but
> still gets one record per call, so it is capped by the 5 calls/s per shard
> quota.
> The default of 1 dates from the first contribution of the component
> (72370df11c97, 2015), with no stated reason. The option is documented as
> "Maximum number of records that will be fetched in each poll" with default 1,
> and the consumer does what that says. That is why this is filed as an
> Improvement (a default change) and not as a Bug.
> Secondary point (read from the source): {{Kinesis2Consumer.java}} lines
> 139-142 go over the shards with
> {{getCurrentShardList().parallelStream().forEach(...)}}. That runs the
> blocking {{GetShardIterator}}/{{GetRecords}} calls, and also the routing of
> the records ({{processBatch}} is called from the same lambda), on the
> JVM-wide {{ForkJoinPool.commonPool()}}. Every other parallel stream and
> default-async {{CompletableFuture}} in the JVM shares that pool, and its
> parallelism (CPU count - 1) cannot be configured per endpoint.
> h3. How to reproduce
> These numbers are measured. The route is
> {{from("aws2-kinesis://stream?amazonKinesisClient=#client&initialDelay=0" +
> options)}}, run with a Mockito {{KinesisClient}} that holds one shard of 1000
> records. {{GetRecords}} returns {{min(Limit, remaining)}} records and records
> the Limit and the calling thread.
> ||options||records routed||GetRecords calls||Limit sent||elapsed||
> |(defaults)|5|5|1|2.5 s, with 995 records still waiting|
> |{{greedy=true}} (50-record shard)|50|51|1| |
> |{{maxResultsPerRequest=1000}} (control)|1000|1|1000|0.65 s, startup included|
> With 8 shards, {{GetRecords}} ran on {{ForkJoinPool.commonPool-worker-1}} to
> {{-7}} and on the consumer's scheduler thread.
> h3. Proposed change
> * Raise the default of {{maxResultsPerRequest}}, for example to 1000, or to
> the API maximum of 10,000.
> ** One response per shard is held in memory while its records are routed one
> by one.
> ** At 10,000 that response can reach 10 MiB per shard per call, and a shard
> serves at most 2 MiB/s of reads.
> ** 1000 keeps the memory of a poll smaller and still gives up to 1000x more
> records at the same call rate (an estimate, assuming the shard has the
> records).
> * Document how {{maxResultsPerRequest}}, {{delay}} and {{greedy}} relate to
> the per-shard quotas (5 GetRecords per second, 2 MiB per second).
> * Optionally, log a warning while {{MillisBehindLatest}} keeps growing.
> * Secondary: run the shard fetches on an executor managed by Camel
> ({{ExecutorServiceManager}}, sized by an option), or one shard after the
> other, rather than on the common pool.
> h3. Impact / upgrade
> * The default changes, so this needs a {{camel-aws2-kinesis}} entry in the
> 4.24 upgrade guide: each poll returns more records, and
> {{maxResultsPerRequest=1}} keeps the old behaviour.
> * Only the polling consumer is affected; the KCL consumer
> ({{useKclConsumers=true}}) does not use this option.
> * The {{aws-kinesis-source}} Kamelet (camel-kamelets 4.23.0) does not set
> {{maxResultsPerRequest}}, so it would pick up the new default.
> * The same code is on camel-4.18.5, camel-4.22.2 and camel-4.23.0.
> h3. Related
> * CAMEL-24156 and CAMEL-24157 (umbrellas of the July 2026 review) fixed other
> problems of the Kinesis consumer, not the default limit.
> * CAMEL-20958: the logs attached there already show the consumer running on
> {{ForkJoinPool.commonPool}} threads.
> * CAMEL-26599: the KCL consumer, which this change does not touch.
> _Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
