You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/18 05:28:55 UTC
(camel) branch main updated: CAMEL-24197: camel-aws2-kinesis - Fix fixed-shardId DescribeStream on every poll and batch partition key (#24878)
This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 48a28da60ef1 CAMEL-24197: camel-aws2-kinesis - Fix fixed-shardId DescribeStream on every poll and batch partition key (#24878)
48a28da60ef1 is described below
commit 48a28da60ef1ad6cdfee4a80bdb3eb93a4ba5a26
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Sat Jul 18 07:28:48 2026 +0200
CAMEL-24197: camel-aws2-kinesis - Fix fixed-shardId DescribeStream on every poll and batch partition key (#24878)
CAMEL-24197: camel-aws2-kinesis - Fix fixed-shardId DescribeStream on every poll and batch partition key
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <da...@apache.org>
---
.../camel/catalog/components/aws2-kinesis.json | 5 +-
.../camel/component/aws2/kinesis/aws2-kinesis.json | 5 +-
.../component/aws2/kinesis/Kinesis2Constants.java | 8 +
.../component/aws2/kinesis/Kinesis2Consumer.java | 108 ++++++--------
.../component/aws2/kinesis/Kinesis2Producer.java | 26 +++-
.../aws2/kinesis/Kinesis2ProducerTest.java | 166 +++++++++++++++++++++
.../KinesisConsumerClosedShardWithSilentTest.java | 17 +--
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 20 +++
.../dsl/Kinesis2EndpointBuilderFactory.java | 16 ++
9 files changed, 290 insertions(+), 81 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
index d804f6e48420..a6ebc21581e0 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
@@ -68,8 +68,9 @@
"CamelMessageTimestamp": { "index": 3, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The timestamp of the message", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#MESSAGE_TIMESTAMP" },
"CamelKinesisDbResumeAction": { "index": 4, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The resume action to execute when resuming.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RESUME_ACTION" },
"CamelAwsKinesisShardId": { "index": 5, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The shard ID of the shard where the data record was placed.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#SHARD_ID" },
- "CamelAwsKinesisFailedRecordCount": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of records that failed in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#FAILED_RECORD_COUNT" },
- "CamelAwsKinesisRecordCount": { "index": 7, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The total number of records in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RECORD_COUNT" }
+ "CamelAwsKinesisPartitionKeys": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "List<String>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "A list of partition keys for batch operations. When set, each record in the batch is assigned the partition key at the corresponding index, allowing records to be routed to different shards. If not set, the single Camel [...]
+ "CamelAwsKinesisFailedRecordCount": { "index": 7, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of records that failed in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#FAILED_RECORD_COUNT" },
+ "CamelAwsKinesisRecordCount": { "index": 8, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The total number of records in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RECORD_COUNT" }
},
"properties": {
"streamName": { "index": 0, "kind": "path", "displayName": "Stream Name", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Name of the stream" },
diff --git a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
index d804f6e48420..a6ebc21581e0 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
+++ b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
@@ -68,8 +68,9 @@
"CamelMessageTimestamp": { "index": 3, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The timestamp of the message", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#MESSAGE_TIMESTAMP" },
"CamelKinesisDbResumeAction": { "index": 4, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The resume action to execute when resuming.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RESUME_ACTION" },
"CamelAwsKinesisShardId": { "index": 5, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The shard ID of the shard where the data record was placed.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#SHARD_ID" },
- "CamelAwsKinesisFailedRecordCount": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of records that failed in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#FAILED_RECORD_COUNT" },
- "CamelAwsKinesisRecordCount": { "index": 7, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The total number of records in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RECORD_COUNT" }
+ "CamelAwsKinesisPartitionKeys": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "List<String>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "A list of partition keys for batch operations. When set, each record in the batch is assigned the partition key at the corresponding index, allowing records to be routed to different shards. If not set, the single Camel [...]
+ "CamelAwsKinesisFailedRecordCount": { "index": 7, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of records that failed in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#FAILED_RECORD_COUNT" },
+ "CamelAwsKinesisRecordCount": { "index": 8, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The total number of records in a batch put operation.", "constantName": "org.apache.camel.component.aws2.kinesis.Kinesis2Constants#RECORD_COUNT" }
},
"properties": {
"streamName": { "index": 0, "kind": "path", "displayName": "Stream Name", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Name of the stream" },
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Constants.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Constants.java
index 5473cef5a58f..f8cbb3473238 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Constants.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Constants.java
@@ -41,6 +41,14 @@ public interface Kinesis2Constants {
@Metadata(description = "The shard ID of the shard where the data record was placed.", javaType = "String")
String SHARD_ID = "CamelAwsKinesisShardId";
+ @Metadata(label = "producer",
+ description = "A list of partition keys for batch operations. When set, each record in the batch "
+ + "is assigned the partition key at the corresponding index, allowing records to be routed "
+ + "to different shards. If not set, the single CamelAwsKinesisPartitionKey header is used "
+ + "for all records.",
+ javaType = "List<String>")
+ String PARTITION_KEYS = "CamelAwsKinesisPartitionKeys";
+
// Batch operation response metadata
@Metadata(label = "producer",
description = "The number of records that failed in a batch put operation.", javaType = "Integer")
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 c08263fd6b13..c8ce34fa8dd3 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,6 +19,7 @@ package org.apache.camel.component.aws2.kinesis;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.ArrayDeque;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
@@ -42,13 +43,12 @@ import org.apache.camel.util.CastUtils;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
-import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
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.Shard;
import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
@@ -105,40 +105,19 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
return 0;
}
- var request = DescribeStreamRequest
- .builder()
- .streamName(getEndpoint().getConfiguration().getStreamName())
- .build();
- DescribeStreamResponse response;
- if (getEndpoint().getConfiguration().isAsyncClient()) {
- try {
- response = connection
- .getAsyncClient(getEndpoint())
- .describeStream(request)
- .get();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return 0;
- } catch (ExecutionException e) {
- throw new RuntimeException(e);
- }
- } else {
- response = connection
- .getClient(getEndpoint())
- .describeStream(request);
- }
-
- var shard = response
- .streamDescription()
- .shards()
- .stream()
- .filter(shardItem -> shardItem
- .shardId()
- .equalsIgnoreCase(getEndpoint()
- .getConfiguration()
- .getShardId()))
+ // Use the cached shard list from ShardMonitor instead of calling DescribeStream
+ // on every poll. ShardMonitor uses ListShards (paginated, 100 TPS limit) vs
+ // DescribeStream (unpaginated, 10 TPS limit).
+ Shard shard = getCurrentShardList().stream()
+ .filter(s -> s.shardId().equalsIgnoreCase(shardId))
.findFirst()
- .orElseThrow(() -> new IllegalStateException("The shard can't be found"));
+ .orElse(null);
+
+ if (shard == null) {
+ LOG.warn("Configured shardId {} not found in shard list for stream {}",
+ shardId, getEndpoint().getConfiguration().getStreamName());
+ return 0;
+ }
fetchAndPrepareRecordsForCamel(shard, connection, processedExchangeCount);
@@ -471,33 +450,42 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
}
private List<Shard> getShardList(final KinesisConnection kinesisConnection) {
- var request = ListShardsRequest
- .builder()
- .streamName(getEndpoint().getConfiguration().getStreamName())
- .build();
-
- List<Shard> shardList;
- if (getEndpoint().getConfiguration().isAsyncClient()) {
- try {
- shardList = kinesisConnection
- .getAsyncClient(getEndpoint())
- .listShards(request)
- .get()
- .shards();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } catch (ExecutionException e) {
- throw new RuntimeException(e);
+ List<Shard> allShards = new ArrayList<>();
+ String nextToken = null;
+
+ do {
+ ListShardsRequest.Builder requestBuilder = ListShardsRequest.builder();
+ if (nextToken != null) {
+ requestBuilder.nextToken(nextToken);
+ } else {
+ requestBuilder.streamName(getEndpoint().getConfiguration().getStreamName());
}
- } else {
- shardList = kinesisConnection
- .getClient(getEndpoint())
- .listShards(request)
- .shards();
- }
+ ListShardsRequest request = requestBuilder.build();
+
+ ListShardsResponse response;
+ if (getEndpoint().getConfiguration().isAsyncClient()) {
+ try {
+ response = kinesisConnection
+ .getAsyncClient(getEndpoint())
+ .listShards(request)
+ .get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ response = kinesisConnection
+ .getClient(getEndpoint())
+ .listShards(request);
+ }
+
+ allShards.addAll(response.shards());
+ nextToken = response.nextToken();
+ } while (nextToken != null);
- return shardList;
+ return allShards;
}
}
}
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
index 54a8f884cadc..0858b1d022f2 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
@@ -73,10 +73,17 @@ public class Kinesis2Producer extends DefaultProducer {
}
}
+ @SuppressWarnings("unchecked")
private void sendBatchRecords(Exchange exchange) {
+ List<String> partitionKeys = exchange.getIn().getHeader(Kinesis2Constants.PARTITION_KEYS, List.class);
Object partitionKey = exchange.getIn().getHeader(Kinesis2Constants.PARTITION_KEY);
- ensurePartitionKeyNotNull(partitionKey);
- List<List<PutRecordsRequestEntry>> requestBatchList = createRequestBatchList(exchange, partitionKey);
+ if (partitionKeys == null) {
+ ensurePartitionKeyNotNull(partitionKey);
+ } else if (partitionKeys.isEmpty()) {
+ throw new IllegalArgumentException("CamelAwsKinesisPartitionKeys header must not be empty");
+ }
+ List<List<PutRecordsRequestEntry>> requestBatchList
+ = createRequestBatchList(exchange, partitionKey, partitionKeys);
int totalRecordCount = 0;
int totalFailedCount = 0;
for (List<PutRecordsRequestEntry> requestBatch : requestBatchList) {
@@ -98,9 +105,11 @@ public class Kinesis2Producer extends DefaultProducer {
message.setHeader(Kinesis2Constants.FAILED_RECORD_COUNT, totalFailedCount);
}
- private List<List<PutRecordsRequestEntry>> createRequestBatchList(Exchange exchange, Object partitionKey) {
+ private List<List<PutRecordsRequestEntry>> createRequestBatchList(
+ Exchange exchange, Object partitionKey, List<String> partitionKeys) {
List<List<PutRecordsRequestEntry>> requestBatchList = new ArrayList<>();
List<PutRecordsRequestEntry> requestBatch = new ArrayList<>(MAX_BATCH_SIZE);
+ int index = 0;
for (Object record : exchange.getIn().getBody(Iterable.class)) {
SdkBytes sdkBytes;
if (record instanceof byte[] bytes) {
@@ -116,15 +125,24 @@ public class Kinesis2Producer extends DefaultProducer {
"Record type not supported. Must be byte[], ByteBuffer, InputStream or UTF-8 String");
}
+ String key;
+ if (partitionKeys != null && index < partitionKeys.size()) {
+ key = partitionKeys.get(index);
+ } else {
+ ensurePartitionKeyNotNull(partitionKey);
+ key = partitionKey.toString();
+ }
+
PutRecordsRequestEntry putRecordsRequestEntry = PutRecordsRequestEntry.builder()
.data(sdkBytes)
- .partitionKey(partitionKey.toString())
+ .partitionKey(key)
.build();
requestBatch.add(putRecordsRequestEntry);
if (requestBatch.size() == MAX_BATCH_SIZE) {
requestBatchList.add(requestBatch);
requestBatch = new ArrayList<>(MAX_BATCH_SIZE);
}
+ index++;
}
if (!requestBatch.isEmpty()) {
requestBatchList.add(requestBatch);
diff --git a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java
new file mode 100644
index 000000000000..4282e96a320b
--- /dev/null
+++ b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.util.Arrays;
+import java.util.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest;
+import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry;
+import software.amazon.awssdk.services.kinesis.model.PutRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.PutRecordsResultEntry;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class Kinesis2ProducerTest {
+
+ @Mock
+ private KinesisClient kinesisClient;
+
+ private final CamelContext context = new DefaultCamelContext();
+ private final Kinesis2Component component = new Kinesis2Component(context);
+ private Kinesis2Producer producer;
+
+ @BeforeEach
+ public void setup() {
+ component.start();
+
+ Kinesis2Configuration configuration = new Kinesis2Configuration();
+ configuration.setAmazonKinesisClient(kinesisClient);
+ configuration.setStreamName("testStream");
+
+ Kinesis2Endpoint endpoint = new Kinesis2Endpoint("aws2-kinesis:testStream", configuration, component);
+ endpoint.start();
+ producer = new Kinesis2Producer(endpoint);
+ producer.setConnection(component.getConnection());
+ producer.start();
+ }
+
+ @Test
+ public void batchWithSinglePartitionKeyAppliesSameKeyToAll() throws Exception {
+ when(kinesisClient.putRecords(any(PutRecordsRequest.class)))
+ .thenReturn(PutRecordsResponse.builder()
+ .failedRecordCount(0)
+ .records(PutRecordsResultEntry.builder().build(),
+ PutRecordsResultEntry.builder().build())
+ .build());
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "shared-key");
+ exchange.getIn().setBody(Arrays.asList("record1", "record2"));
+
+ producer.process(exchange);
+
+ ArgumentCaptor<PutRecordsRequest> captor = ArgumentCaptor.forClass(PutRecordsRequest.class);
+ verify(kinesisClient).putRecords(captor.capture());
+
+ List<PutRecordsRequestEntry> entries = captor.getValue().records();
+ assertEquals(2, entries.size());
+ assertEquals("shared-key", entries.get(0).partitionKey());
+ assertEquals("shared-key", entries.get(1).partitionKey());
+ }
+
+ @Test
+ public void batchWithPerRecordPartitionKeys() throws Exception {
+ when(kinesisClient.putRecords(any(PutRecordsRequest.class)))
+ .thenReturn(PutRecordsResponse.builder()
+ .failedRecordCount(0)
+ .records(PutRecordsResultEntry.builder().build(),
+ PutRecordsResultEntry.builder().build(),
+ PutRecordsResultEntry.builder().build())
+ .build());
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEYS,
+ Arrays.asList("key-A", "key-B", "key-C"));
+ exchange.getIn().setBody(Arrays.asList("record1", "record2", "record3"));
+
+ producer.process(exchange);
+
+ ArgumentCaptor<PutRecordsRequest> captor = ArgumentCaptor.forClass(PutRecordsRequest.class);
+ verify(kinesisClient).putRecords(captor.capture());
+
+ List<PutRecordsRequestEntry> entries = captor.getValue().records();
+ assertEquals(3, entries.size());
+ assertEquals("key-A", entries.get(0).partitionKey());
+ assertEquals("key-B", entries.get(1).partitionKey());
+ assertEquals("key-C", entries.get(2).partitionKey());
+ }
+
+ @Test
+ public void batchWithFewerPartitionKeysFallsBackToSingleKey() throws Exception {
+ when(kinesisClient.putRecords(any(PutRecordsRequest.class)))
+ .thenReturn(PutRecordsResponse.builder()
+ .failedRecordCount(0)
+ .records(PutRecordsResultEntry.builder().build(),
+ PutRecordsResultEntry.builder().build(),
+ PutRecordsResultEntry.builder().build())
+ .build());
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "fallback-key");
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEYS,
+ Arrays.asList("key-A"));
+ exchange.getIn().setBody(Arrays.asList("record1", "record2", "record3"));
+
+ producer.process(exchange);
+
+ ArgumentCaptor<PutRecordsRequest> captor = ArgumentCaptor.forClass(PutRecordsRequest.class);
+ verify(kinesisClient).putRecords(captor.capture());
+
+ List<PutRecordsRequestEntry> entries = captor.getValue().records();
+ assertEquals(3, entries.size());
+ assertEquals("key-A", entries.get(0).partitionKey());
+ assertEquals("fallback-key", entries.get(1).partitionKey());
+ assertEquals("fallback-key", entries.get(2).partitionKey());
+ }
+
+ @Test
+ public void batchWithFewerPartitionKeysAndNoFallbackThrows() {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEYS,
+ Arrays.asList("key-A"));
+ exchange.getIn().setBody(Arrays.asList("record1", "record2", "record3"));
+
+ assertThrows(IllegalArgumentException.class, () -> producer.process(exchange));
+ }
+
+ @Test
+ public void batchWithEmptyPartitionKeysListThrows() {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEYS, List.of());
+ exchange.getIn().setBody(Arrays.asList("record1"));
+
+ assertThrows(IllegalArgumentException.class, () -> producer.process(exchange));
+ }
+}
diff --git a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
index eb438c82f72f..b1f996362a64 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
@@ -35,7 +35,6 @@ 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.DescribeStreamRequest;
-import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
@@ -46,12 +45,12 @@ 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 software.amazon.awssdk.services.kinesis.model.StreamDescription;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -127,21 +126,13 @@ public class KinesisConsumerClosedShardWithSilentTest {
}
@Test
- public void itDoesNotMakeADescribeStreamRequestIfShardIdIsSet() throws Exception {
-
- SequenceNumberRange range = SequenceNumberRange.builder().endingSequenceNumber("20").build();
- Shard shard = Shard.builder().shardId("shardId").sequenceNumberRange(range).build();
- ArrayList<Shard> shardList = new ArrayList<>();
- shardList.add(shard);
-
- when(kinesisClient.describeStream(any(DescribeStreamRequest.class)))
- .thenReturn(DescribeStreamResponse.builder()
- .streamDescription(StreamDescription.builder().shards(shardList).build()).build());
-
+ public void itUsesShardMonitorCacheNotDescribeStreamForFixedShardId() throws Exception {
underTest.getEndpoint().getConfiguration().setShardId("shardId");
underTest.poll();
+ verify(kinesisClient, never()).describeStream(any(DescribeStreamRequest.class));
+
final ArgumentCaptor<GetShardIteratorRequest> getShardIteratorReqCap
= ArgumentCaptor.forClass(GetShardIteratorRequest.class);
diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 88c5a3f180c4..2d79112cac9f 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -414,6 +414,26 @@ The `camel infra list` table now sizes its `DESCRIPTION` column to the terminal
the `IMPLEMENTATION` and `SERVICE_DATA` columns with an ellipsis instead of letting the raw service
data overflow the terminal. The complete, structured service data remains available via `--json`.
+=== camel-aws2-kinesis - Fixed-shardId consumer no longer calls DescribeStream on every poll
+
+The Kinesis consumer with a configured `shardId` previously called the `DescribeStream` API on every
+poll cycle to locate the shard, risking AWS rate limits (10 TPS per account) and failing for streams
+with more than 100 shards (no pagination). The consumer now uses the same cached shard list from the
+`ShardMonitor` background thread that the multi-shard path already uses. The `ShardMonitor` uses the
+`ListShards` API (paginated, 100 TPS limit) and now paginates through all pages, supporting streams
+with any number of shards.
+
+This is a transparent performance improvement with no configuration changes required.
+
+=== camel-aws2-kinesis - Per-record partition keys in batch producer
+
+The batch producer (body is an `Iterable`) previously applied a single `CamelAwsKinesisPartitionKey`
+header to all records, routing them all to the same shard. A new `CamelAwsKinesisPartitionKeys` header
+(`List<String>`) is now supported: when set, each record in the batch is assigned the partition key at
+the corresponding index. If the list has fewer entries than records, the remaining records fall back to
+the single `CamelAwsKinesisPartitionKey` header. The existing single-key behavior is unchanged when
+the new header is not set.
+
=== camel-azure-storage-blob / camel-azure-storage-datalake - download contained within fileDir
When `fileDir` is configured, the Azure Storage Blob and DataLake consumers now ensure the downloaded
diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java
index f5ee78963c33..1c731d648672 100644
--- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java
+++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java
@@ -2889,6 +2889,22 @@ public interface Kinesis2EndpointBuilderFactory {
public String awsKinesisShardId() {
return "CamelAwsKinesisShardId";
}
+ /**
+ * A list of partition keys for batch operations. When set, each record
+ * in the batch is assigned the partition key at the corresponding
+ * index, allowing records to be routed to different shards. If not set,
+ * the single CamelAwsKinesisPartitionKey header is used for all
+ * records.
+ *
+ * The option is a: {@code List<String>} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code AwsKinesisPartitionKeys}.
+ */
+ public String awsKinesisPartitionKeys() {
+ return "CamelAwsKinesisPartitionKeys";
+ }
/**
* The number of records that failed in a batch put operation.
*