oscerd opened a new pull request, #24717: URL: https://github.com/apache/camel/pull/24717
## Description Fixes [CAMEL-24080](https://issues.apache.org/jira/browse/CAMEL-24080). When no `shardId` is configured, `Kinesis2Consumer.poll()` iterates the shard list with a `parallelStream()` and each worker runs `fetchAndPrepareRecordsForCamel(...)` concurrently. The shared per-consumer state written from those workers was not thread-safe: - **`currentShardIterators`** was a plain `HashMap` (and **`warnLogged`** a plain `HashSet`), mutated concurrently via `updateShardIterator(...)` / `getShardIterator(...)`. Concurrent `put`/`add` can corrupt the map or lose iterator updates, leading to **lost or duplicated records** when consuming a stream with more than one shard. - **`processedExchangeCount`** was updated with `getAndSet(processBatch(...))` (and reset with `set(0)` on the closed-shard path), so parallel shards overwrote each other and `poll()` returned only the **last-finishing shard's count** instead of the total. That count feeds `ScheduledBatchPollingConsumer` idle/backoff accounting (e.g. `backoffIdleThreshold`). ## Fix - Use `ConcurrentHashMap` for `currentShardIterators` and `ConcurrentHashMap.newKeySet()` for `warnLogged`. - Accumulate the poll count with `addAndGet(...)` and drop the `set(0)` reset on the closed-shard branch. - `ConcurrentHashMap` forbids `null` values, and the closed-shard detection (`isShardClosed`, `updateShardIterator`) relied on storing a `null` iterator to mark a shard closed. A sentinel (empty string — already treated as "no iterator" by the existing `ObjectHelper.isEmpty(...)` checks) is stored instead, so closed-shard behavior is unchanged. ## Tests - New `KinesisConsumerMultiShardTest` — two open shards, two records each; asserts `poll()` returns **4** (the sum), which fails against the old `getAndSet` logic (returned 2). - Existing `KinesisConsumerClosedShardWith{Silent,Fail}Test` continue to pass, covering the null→sentinel closed-shard path. ## Backport The same code is present on `camel-4.18.x` and `camel-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_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
