Andrea Cosentino created CAMEL-24080:
----------------------------------------
Summary: camel-aws2-kinesis: consumer shard state is not
thread-safe when consuming multiple shards
Key: CAMEL-24080
URL: https://issues.apache.org/jira/browse/CAMEL-24080
Project: Camel
Issue Type: Bug
Components: camel-aws2-kinesis
Reporter: Andrea Cosentino
Assignee: Andrea Cosentino
Fix For: 4.14.9, 4.22.0, 4.18.4
When no explicit shardId is configured, Kinesis2Consumer.poll() iterates the
shard list with parallelStream() and each worker runs
fetchAndPrepareRecordsForCamel(...) concurrently. The shared per-consumer state
written from those workers is not thread-safe:
- {{currentShardIterators}} is a plain {{java.util.HashMap}} mutated
concurrently via {{updateShardIterator(...)}} and {{getShardIterator(...)}} —
concurrent puts can silently corrupt the map / lose iterator updates, leading
to lost or duplicated records when consuming multiple shards.
- {{warnLogged}} is a plain {{HashSet}} with the same problem.
- {{processedExchangeCount}} is updated with {{getAndSet(processBatch(...))}}
and {{set(0)}} per shard, so parallel shards overwrite each other and
{{poll()}} returns only the last-finishing shard's count instead of the total —
breaking ScheduledBatchPollingConsumer idle/backoff accounting (e.g.
backoffIdleThreshold).
Code references (main): Kinesis2Consumer.java field declarations (~line 66-67),
parallelStream at ~line 142-143, getAndSet/set(0) at ~line 168/201,
updateShardIterator at ~line 213-215.
Fix direction: use ConcurrentHashMap / a concurrent set for the shared shard
state and accumulate the poll count with addAndGet(). The same code is present
on camel-4.18.x and camel-4.14.x, so the fix should be backported to both.
Related (broader feature ask, not a duplicate): CAMEL-19582.
Found during an agent-assisted audit of the AWS components; verified by reading
the code at the cited lines.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
