Federico Mariani created CAMEL-26670:
----------------------------------------
Summary: camel-aws2-kinesis - the polling consumer reads the
shards and routes their records on ForkJoinPool.commonPool()
Key: CAMEL-26670
URL: https://issues.apache.org/jira/browse/CAMEL-26670
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 ({{useKclConsumers=false}}) reads the
shards of a stream with {{getCurrentShardList().parallelStream().forEach(shard
-> fetchAndPrepareRecordsForCamel(...))}}
({{components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java}}
lines 140-142 on main 7978db3d42dd). The lambda does the blocking
{{GetShardIterator}} and {{GetRecords}} calls (with {{asyncClient=true}} it
blocks on {{.get()}}, lines 216-221), and it routes the records:
{{processBatch}} (line 193) calls the route for each record on the same thread.
A parallel stream runs on {{ForkJoinPool.commonPool()}} and on the calling
thread (the consumer's scheduler thread). As a result:
* the routes run on JVM-wide threads named
{{ForkJoinPool.commonPool-worker-N}}, which the {{ExecutorServiceManager}} does
not manage: no thread pool profile, no Camel thread name, not shut down with
the consumer.
* the number of shards read at the same time is the common-pool parallelism +
1. That parallelism is {{availableProcessors() - 1}}, at least 1. It cannot be
set per endpoint, only for the whole JVM
({{java.util.concurrent.ForkJoinPool.common.parallelism}}).
* other parallel streams, {{CompletableFuture.*Async}} calls without an
executor, and every other {{aws2-kinesis}} consumer of the JVM share these
threads, and a slow route holds them for the whole poll. Since CAMEL-26650 a
poll routes up to 1000 records per shard there (before, 1).
* on a forced shutdown only the scheduler thread is interrupted. The
common-pool workers keep going through their response, and each of the
remaining records fails with {{RouteStoppingException}}.
h3. How to reproduce
Measured on main + PR #29272 (JDK 25, 10 CPUs) with a Mockito
{{KinesisClient}}: 4 shards of 5000 records each, {{GetRecords}} returns
{{min(Limit, remaining)}} records. The route
{{from("aws2-kinesis://stream?amazonKinesisClient=#kinesisClient").process(<record
the thread name>).delay(20).syncDelayed()}} was stopped after 5 seconds:
* 620 of the 827 records that entered the route ran on
{{ForkJoinPool.commonPool-worker-N}}, and 207 on {{Camel (camel-2) thread #4 -
aws2-kinesis://stream}}, the scheduler thread.
* with one shard, all records ran on the scheduler thread, because a
one-element parallel stream does not fork.
* the forced stop interrupted one record: stopping the scheduler cancels its
task with an interrupt ({{DefaultScheduledPollConsumerScheduler.doStop}}),
which reaches the scheduler thread only. The other 3176 records left in the
responses failed with {{RouteStoppingException}}.
{{ForkJoinPool.getCommonPoolParallelism()}} with {{-XX:ActiveProcessorCount}}:
||CPUs||1||2||4||10||
||common-pool parallelism|1|1|3|9|
The logs attached to CAMEL-20958 (2024) show the consumer on
{{ForkJoinPool.commonPool-worker-3}} and {{-4}} too.
h3. Proposed change
* Read the shards on an executor that Camel manages: create it in {{doStart}}
with {{getCamelContext().getExecutorServiceManager()}} (named, for example,
after the existing {{Kinesis_shard_monitor}}), and shut it down in {{doStop}}.
Submit one task per shard, and let {{poll()}} wait for all of them, so that a
poll still sends one {{GetRecords}} per shard and returns the count of routed
records.
* Size it with a new endpoint option. The default needs a decision: the current
effective concurrency depends on the CPU count (see the table), one shard after
the other is the simplest but slower for streams with many shards, and a fixed
number gives the same behaviour on every machine.
* Points to watch:
** an exception of one shard ({{shardClosed=fail}}, the {{RuntimeException}}
wrappers in {{fetchAndPrepareRecordsForCamel}}) fails the poll today. The
executor version must wait for all tasks, then rethrow the first cause
unwrapped, so that the poll strategy, the error counters and the back-off
behave as now.
** stopping the executor interrupts the shard tasks, which today happens only
to the scheduler thread. This works best together with the follow-up that makes
{{processBatch}} stop when {{isBatchAllowed()}} is false.
** the shard state is already thread-safe (CAMEL-24080), and each exchange
still runs on one thread, so thread-bound state (MDC, transactions) behaves as
now.
* A configured {{shardId}} (lines 115-137) does not use the parallel stream and
is not affected.
h3. Impact / upgrade
* The routes run on Camel threads instead of
{{ForkJoinPool.commonPool-worker-N}}. If the default concurrency differs from
today's, the throughput of multi-shard streams changes too. Both need a 4.24
(or later) upgrade-guide entry.
* Only the polling consumer is affected. The KCL consumer
({{KclKinesis2Consumer}}) extends {{DefaultConsumer}} and does not use this
code.
* The same {{parallelStream()}} is in camel-4.18.5, camel-4.22.2 and
camel-4.23.0. It dates from CAMEL-19583 (commit f8bad1bbf268, 4.0.0), which
added reading all shards in one route; no reason for the common pool is given
there.
h3. Related
* CAMEL-26650 (PR #29272): the default of 1000 records per shard per poll. Its
description lists the common pool as a secondary point, and the PR leaves it
for this follow-up.
* CAMEL-24080 (commit 75695dfe2418): made the shard state thread-safe for the
parallel stream.
* CAMEL-20958: its logs show the common-pool threads.
* The follow-up about {{processBatch}} and {{isBatchAllowed()}}, filed at the
same time.
_Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
