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 <[email protected]>
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 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../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.
*
