This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 75695dfe2418 CAMEL-24080: camel-aws2-kinesis - make consumer shard
state thread-safe for multi-shard streams (#24717)
75695dfe2418 is described below
commit 75695dfe24182b6b5ffcf23a0d4545e67ff89801
Author: Andrea Cosentino <[email protected]>
AuthorDate: Wed Jul 15 15:36:53 2026 +0200
CAMEL-24080: camel-aws2-kinesis - make consumer shard state thread-safe for
multi-shard streams (#24717)
Use ConcurrentHashMap / ConcurrentHashMap.newKeySet() for the shard state
written from poll()'s parallelStream, and accumulate the poll count with
addAndGet(). A sentinel replaces the null closed-shard marker
(ConcurrentHashMap forbids null values). Adds KinesisConsumerMultiShardTest.
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../component/aws2/kinesis/Kinesis2Consumer.java | 24 +++--
.../kinesis/KinesisConsumerMultiShardTest.java | 119 +++++++++++++++++++++
2 files changed, 134 insertions(+), 9 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
index 6f836492ecef..c08263fd6b13 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -19,12 +19,11 @@ package org.apache.camel.component.aws2.kinesis;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.ArrayDeque;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -60,11 +59,17 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
private static final String UNIX_TIMESTAMP_MILLIS_REGEX = "^\\d{1,13}$";
private static final String UNIX_TIMESTAMP_DOUBLE_REGEX =
"^[-+]?\\d+(\\.\\d+)?([eE][-+]?\\d+)?$";
+ // Sentinel stored in currentShardIterators to mark a shard as closed. A
closed shard has no next
+ // shard iterator (Kinesis returns null); ConcurrentHashMap forbids null
values, so an empty
+ // string is used instead and is treated as "no iterator" by the
ObjectHelper.isEmpty checks.
+ private static final String CLOSED_SHARD_ITERATOR = "";
+
private KinesisConnection connection;
private ResumeStrategy resumeStrategy;
- private final Map<String, String> currentShardIterators = new HashMap<>();
- private final Set<String> warnLogged = new HashSet<>();
+ // Written concurrently from poll()'s parallelStream over the shard list,
so both must be thread-safe.
+ private final Map<String, String> currentShardIterators = new
ConcurrentHashMap<>();
+ private final Set<String> warnLogged = ConcurrentHashMap.newKeySet();
private volatile List<Shard> currentShardList = List.of();
private static final String SHARD_MONITOR_EXECUTOR_NAME =
"Kinesis_shard_monitor";
@@ -84,7 +89,7 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
}
public boolean isShardClosed(String shardId) {
- return currentShardIterators.get(shardId) == null &&
currentShardIterators.containsKey(shardId);
+ return ObjectHelper.isEmpty(currentShardIterators.get(shardId)) &&
currentShardIterators.containsKey(shardId);
}
@Override
@@ -164,8 +169,7 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
}
if (ObjectHelper.isEmpty(shardIterator)) {
- // Unable to get an interator so shard must be closed
- processedExchangeCount.set(0);
+ // Unable to get an iterator so shard must be closed; contribute
nothing to the poll count
return;
}
@@ -198,7 +202,7 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
try {
Queue<Exchange> exchanges = createExchanges(shard,
result.records());
-
processedExchangeCount.getAndSet(processBatch(CastUtils.cast(exchanges)));
+
processedExchangeCount.addAndGet(processBatch(CastUtils.cast(exchanges)));
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -211,7 +215,9 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
}
private void updateShardIterator(Shard shard, String nextShardIterator) {
- currentShardIterators.put(shard.shardId(), nextShardIterator);
+ // A null next shard iterator means the shard is now closed; store the
sentinel rather than
+ // null (ConcurrentHashMap forbids null values) so isShardClosed keeps
detecting it.
+ currentShardIterators.put(shard.shardId(), nextShardIterator == null ?
CLOSED_SHARD_ITERATOR : nextShardIterator);
}
@Override
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerMultiShardTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerMultiShardTest.java
new file mode 100644
index 000000000000..64184b641ad1
--- /dev/null
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerMultiShardTest.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
+import software.amazon.awssdk.services.kinesis.model.ListShardsRequest;
+import software.amazon.awssdk.services.kinesis.model.ListShardsResponse;
+import software.amazon.awssdk.services.kinesis.model.Record;
+import software.amazon.awssdk.services.kinesis.model.SequenceNumberRange;
+import software.amazon.awssdk.services.kinesis.model.Shard;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Verifies that polling a stream with more than one open shard is handled
correctly: every shard is fetched and the
+ * number of processed exchanges returned by {@link Kinesis2Consumer#poll()}
is the sum across all shards rather than
+ * the count of whichever shard happened to finish last.
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class KinesisConsumerMultiShardTest {
+ @Mock
+ private KinesisClient kinesisClient;
+ @Mock
+ private AsyncProcessor processor;
+
+ private final CamelContext context = new DefaultCamelContext();
+ private final Kinesis2Component component = new Kinesis2Component(context);
+
+ private Kinesis2Consumer underTest;
+
+ @BeforeEach
+ public void setup() {
+ SequenceNumberRange range =
SequenceNumberRange.builder().endingSequenceNumber("2").build();
+ ArrayList<Shard> shardList = new ArrayList<>();
+
shardList.add(Shard.builder().shardId("shardId1").sequenceNumberRange(range).build());
+
shardList.add(Shard.builder().shardId("shardId2").sequenceNumberRange(range).build());
+
+ // Two records per getRecords call, with a non-null next iterator so
both shards stay open across the poll.
+ var records = GetRecordsResponse.builder()
+ .nextShardIterator("nextIterator")
+ .records(
+ Record.builder().sequenceNumber("1")
+ .data(SdkBytes.fromString("Hello",
Charset.defaultCharset()))
+ .build(),
+ Record.builder().sequenceNumber("2")
+ .data(SdkBytes.fromString("Bye",
Charset.defaultCharset()))
+ .build())
+ .build();
+
when(kinesisClient.getRecords(any(GetRecordsRequest.class))).thenReturn(records);
+
when(kinesisClient.getShardIterator(any(GetShardIteratorRequest.class)))
+
.thenReturn(GetShardIteratorResponse.builder().shardIterator("shardIterator").build());
+ when(kinesisClient.listShards(any(ListShardsRequest.class)))
+
.thenReturn(ListShardsResponse.builder().shards(shardList).build());
+
+ component.start();
+
+ Kinesis2Configuration configuration = new Kinesis2Configuration();
+ configuration.setAmazonKinesisClient(kinesisClient);
+ configuration.setIteratorType(ShardIteratorType.LATEST);
+ configuration.setStreamName("streamName");
+
+ Kinesis2Endpoint endpoint = new Kinesis2Endpoint("aws2-kinesis:foo",
configuration, component);
+ endpoint.start();
+ underTest = new Kinesis2Consumer(endpoint, processor);
+ underTest.setConnection(component.getConnection());
+ underTest.start();
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> underTest.getCurrentShardList().size() == 2);
+ }
+
+ @Test
+ public void pollAccumulatesTheRecordCountAcrossAllShards() throws
Exception {
+ // Two shards, two records each -> poll must report all four, not just
the last shard's two.
+ int processed = underTest.poll();
+
+ Assertions.assertEquals(4, processed, "poll() must sum the processed
records across every shard");
+ verify(kinesisClient,
times(2)).getRecords(any(GetRecordsRequest.class));
+ }
+}
