[
https://issues.apache.org/jira/browse/CAMEL-24080?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-24080 started by Andrea Cosentino.
------------------------------------------------
> 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
> Priority: Major
> Fix For: 4.14.9, 4.18.4, 4.22.0
>
>
> 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)
