CAMEL-24080: camel-aws2-kinesis - make consumer shard state thread-safe for multi-shard streams - #24717
Conversation
…fe for multi-shard streams When no shardId is configured, Kinesis2Consumer.poll() iterates the shard list with a parallelStream() and each worker updates shared per-consumer state. That state was not thread-safe: - currentShardIterators was a plain HashMap and warnLogged a plain HashSet, both mutated concurrently -> possible map corruption and lost/duplicated shard iterators when consuming multiple shards. - processedExchangeCount was updated with getAndSet(...) (and reset with set(0) on the closed-shard path), so parallel shards overwrote each other and poll() returned only the last shard's count instead of the total, breaking ScheduledBatchPollingConsumer idle/backoff accounting. Use ConcurrentHashMap / ConcurrentHashMap.newKeySet() for the shared state and accumulate the poll count with addAndGet(). ConcurrentHashMap forbids null values, and the closed-shard detection relied on storing a null iterator, so a sentinel (empty string, already treated as "no iterator" by the existing ObjectHelper.isEmpty checks) is stored instead. Adds KinesisConsumerMultiShardTest asserting poll() sums the records across all shards; existing closed-shard tests continue to pass. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Andrea Cosentino <ancosen@gmail.com>
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Clean, well-targeted fix for a real concurrency bug introduced when synchronized blocks were removed (CAMEL-20199). The analysis is thorough and the changes are minimal:
HashMap/HashSet→ConcurrentHashMap/ConcurrentHashMap.newKeySet()— correct for theparallelStream()mutation inpoll()getAndSet→addAndGet— properly accumulates across shards instead of last-writer-wins- Removing
set(0)on the closed-shard path — prevents resetting counts accumulated by other parallel workers - Empty-string sentinel for null — clean solution for ConcurrentHashMap's null prohibition, logically equivalent to the old null-based detection
The new KinesisConsumerMultiShardTest exercises the exact failure mode (2 shards × 2 records, asserts total is 4). Existing closed-shard tests continue to validate that path.
LGTM.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Well-analyzed concurrency fix that correctly replaces HashMap/HashSet with ConcurrentHashMap equivalents and fixes the processedExchangeCount race from getAndSet to addAndGet.
Key observations:
-
The two bugs are real: (1)
HashMapstructural corruption under concurrentputfromparallelStreamworkers, and (2)getAndSetoverwriting the accumulated count — with 2 shards × 2 records, the old code would report only the last shard's count instead of the sum. -
The sentinel pattern (
CLOSED_SHARD_ITERATOR = "") for null values inConcurrentHashMapis clean.ObjectHelper.isEmpty("")returnstrue, so existing null checks work correctly with the sentinel. -
Removing
processedExchangeCount.set(0)on the closed-shard early-return path is correct — withaddAndGet, a closed shard contributes zero by returning early. The oldset(0)would have been destructive underparallelStreamby zeroing counts from other shards. -
The pre-existing check-then-act in
getShardIteratoris not atomic, but is safe in practice because each shard operates on its own distinct key — no twoparallelStreamworkers contend on the sameshardId. -
The test validates the
addAndGetaccumulation fix (2 shards × 2 records = 4 total). True concurrent contention isn't exercised since Mockito stubs are synchronous, but this is acceptable for a unit test.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Claude Code review on behalf of @gnodet
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 9 tested, 29 compile-only — current: 9 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 9).
|
3 participants
Description
Fixes CAMEL-24080.
When no
shardIdis configured,Kinesis2Consumer.poll()iterates the shard list with aparallelStream()and each worker runsfetchAndPrepareRecordsForCamel(...)concurrently. The shared per-consumer state written from those workers was not thread-safe:currentShardIteratorswas a plainHashMap(andwarnLoggeda plainHashSet), mutated concurrently viaupdateShardIterator(...)/getShardIterator(...). Concurrentput/addcan corrupt the map or lose iterator updates, leading to lost or duplicated records when consuming a stream with more than one shard.processedExchangeCountwas updated withgetAndSet(processBatch(...))(and reset withset(0)on the closed-shard path), so parallel shards overwrote each other andpoll()returned only the last-finishing shard's count instead of the total. That count feedsScheduledBatchPollingConsumeridle/backoff accounting (e.g.backoffIdleThreshold).Fix
ConcurrentHashMapforcurrentShardIteratorsandConcurrentHashMap.newKeySet()forwarnLogged.addAndGet(...)and drop theset(0)reset on the closed-shard branch.ConcurrentHashMapforbidsnullvalues, and the closed-shard detection (isShardClosed,updateShardIterator) relied on storing anulliterator to mark a shard closed. A sentinel (empty string — already treated as "no iterator" by the existingObjectHelper.isEmpty(...)checks) is stored instead, so closed-shard behavior is unchanged.Tests
KinesisConsumerMultiShardTest— two open shards, two records each; assertspoll()returns 4 (the sum), which fails against the oldgetAndSetlogic (returned 2).KinesisConsumerClosedShardWith{Silent,Fail}Testcontinue to pass, covering the null→sentinel closed-shard path.Backport
The same code is present on
camel-4.18.xandcamel-4.14.x; this will be backported to both after merge (fixVersions 4.22.0 / 4.18.4 / 4.14.9).Claude Code on behalf of oscerd