Federico Mariani created CAMEL-26650:
----------------------------------------
Summary: 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.23.0, 4.22.2
Reporter: Federico Mariani
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)
