This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new fd8e8fc38040 CAMEL-24191: camel-aws2-kinesis - fix KCL consumer
session credentials and checkpoint/error handling (#24848)
fd8e8fc38040 is described below
commit fd8e8fc380408966e1d3dbfb4de87e341a8ab621
Author: Andrea Cosentino <[email protected]>
AuthorDate: Fri Jul 17 15:04:49 2026 +0200
CAMEL-24191: camel-aws2-kinesis - fix KCL consumer session credentials and
checkpoint/error handling (#24848)
Backport of #24835 to camel-4.18.x.
Reorder credential branches so session tokens are used (both DynamoDB and
CloudWatch builders); checkpoint only after a successful batch and route
processing failures to the ExceptionHandler without checkpointing so KCL
redelivers. Adds KclKinesis2ConsumerRecordProcessingTest.
Related to CAMEL-24156.
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../aws2/kinesis/KclKinesis2Consumer.java | 47 +++++-----
.../KclKinesis2ConsumerRecordProcessingTest.java | 104 +++++++++++++++++++++
2 files changed, 129 insertions(+), 22 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java
index 8e1a25bda35b..d4b04a2b7d0b 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java
@@ -92,6 +92,12 @@ public class KclKinesis2Consumer extends DefaultConsumer {
if
(ObjectHelper.isEmpty(getEndpoint().getConfiguration().getDynamoDbAsyncClient()))
{
DynamoDbAsyncClientBuilder clientBuilder =
DynamoDbAsyncClient.builder();
if (ObjectHelper.isNotEmpty(configuration.getAccessKey())
+ && ObjectHelper.isNotEmpty(configuration.getSecretKey())
+ &&
ObjectHelper.isNotEmpty(configuration.getSessionToken())) {
+ clientBuilder =
clientBuilder.credentialsProvider(StaticCredentialsProvider
+
.create(AwsSessionCredentials.create(configuration.getAccessKey(),
configuration.getSecretKey(),
+ configuration.getSessionToken())));
+ } else if (ObjectHelper.isNotEmpty(configuration.getAccessKey())
&& ObjectHelper.isNotEmpty(configuration.getSecretKey())) {
clientBuilder =
clientBuilder.credentialsProvider(StaticCredentialsProvider.create(
AwsBasicCredentials.create(configuration.getAccessKey(),
configuration.getSecretKey())));
@@ -99,12 +105,6 @@ public class KclKinesis2Consumer extends DefaultConsumer {
clientBuilder = clientBuilder
.credentialsProvider(
ProfileCredentialsProvider.create(configuration.getProfileCredentialsName()));
- } else if (ObjectHelper.isNotEmpty(configuration.getAccessKey())
- && ObjectHelper.isNotEmpty(configuration.getSecretKey())
- &&
ObjectHelper.isNotEmpty(configuration.getSessionToken())) {
- clientBuilder =
clientBuilder.credentialsProvider(StaticCredentialsProvider
-
.create(AwsSessionCredentials.create(configuration.getAccessKey(),
configuration.getSecretKey(),
- configuration.getSessionToken())));
}
if (ObjectHelper.isNotEmpty(configuration.getRegion())) {
clientBuilder =
clientBuilder.region(Region.of(configuration.getRegion()));
@@ -117,6 +117,12 @@ public class KclKinesis2Consumer extends DefaultConsumer {
if
(ObjectHelper.isEmpty(getEndpoint().getConfiguration().getCloudWatchAsyncClient()))
{
CloudWatchAsyncClientBuilder clientBuilder =
CloudWatchAsyncClient.builder();
if (ObjectHelper.isNotEmpty(configuration.getAccessKey())
+ && ObjectHelper.isNotEmpty(configuration.getSecretKey())
+ &&
ObjectHelper.isNotEmpty(configuration.getSessionToken())) {
+ clientBuilder =
clientBuilder.credentialsProvider(StaticCredentialsProvider
+
.create(AwsSessionCredentials.create(configuration.getAccessKey(),
configuration.getSecretKey(),
+ configuration.getSessionToken())));
+ } else if (ObjectHelper.isNotEmpty(configuration.getAccessKey())
&& ObjectHelper.isNotEmpty(configuration.getSecretKey())) {
clientBuilder =
clientBuilder.credentialsProvider(StaticCredentialsProvider.create(
AwsBasicCredentials.create(configuration.getAccessKey(),
configuration.getSecretKey())));
@@ -124,12 +130,6 @@ public class KclKinesis2Consumer extends DefaultConsumer {
clientBuilder = clientBuilder
.credentialsProvider(
ProfileCredentialsProvider.create(configuration.getProfileCredentialsName()));
- } else if (ObjectHelper.isNotEmpty(configuration.getAccessKey())
- && ObjectHelper.isNotEmpty(configuration.getSecretKey())
- &&
ObjectHelper.isNotEmpty(configuration.getSessionToken())) {
- clientBuilder =
clientBuilder.credentialsProvider(StaticCredentialsProvider
-
.create(AwsSessionCredentials.create(configuration.getAccessKey(),
configuration.getSecretKey(),
- configuration.getSessionToken())));
}
if (ObjectHelper.isNotEmpty(configuration.getRegion())) {
clientBuilder =
clientBuilder.region(Region.of(configuration.getRegion()));
@@ -193,16 +193,19 @@ public class KclKinesis2Consumer extends DefaultConsumer {
public void processRecords(ProcessRecordsInput processRecordsInput) {
try {
LOG.debug("Processing {} record(s)",
processRecordsInput.records().size());
- processRecordsInput.records()
- .forEach(r -> {
- try {
- processor.process(createExchange(r, shardId));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- } catch (Throwable t) {
- LOG.error("Caught throwable while processing records.
Aborting.");
+ for (KinesisClientRecord record :
processRecordsInput.records()) {
+ processor.process(createExchange(record, shardId));
+ }
+ // Checkpoint only after the whole batch has been processed
successfully, so that a
+ // processing failure leaves the records to be redelivered
from the last checkpoint
+ // rather than being silently skipped.
+ processRecordsInput.checkpointer().checkpoint();
+ } catch (ShutdownException | InvalidStateException e) {
+ LOG.warn("Unable to checkpoint after processing Kinesis
records", e);
+ } catch (Exception e) {
+ // Do not checkpoint: let KCL redeliver this batch from the
last successful checkpoint.
+ KclKinesis2Consumer.this.getExceptionHandler()
+ .handleException("Error while processing Kinesis
records; batch will be retried", e);
}
}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KclKinesis2ConsumerRecordProcessingTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KclKinesis2ConsumerRecordProcessingTest.java
new file mode 100644
index 000000000000..5ec14f933b0f
--- /dev/null
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KclKinesis2ConsumerRecordProcessingTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.ByteBuffer;
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.DefaultCamelContext;
+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.services.kinesis.KinesisClient;
+import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
+import software.amazon.kinesis.processor.RecordProcessorCheckpointer;
+import software.amazon.kinesis.retrieval.KinesisClientRecord;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Verifies the KCL consumer only checkpoints after a batch is processed
successfully, so a processing failure leaves
+ * the records to be redelivered rather than silently skipped.
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class KclKinesis2ConsumerRecordProcessingTest {
+
+ @Mock
+ private Processor processor;
+ @Mock
+ private RecordProcessorCheckpointer checkpointer;
+ @Mock
+ private KinesisClientRecord record;
+ @Mock
+ private KinesisClient kinesisClient;
+
+ private final DefaultCamelContext context = new DefaultCamelContext();
+ private KclKinesis2Consumer.CamelKinesisRecordProcessor recordProcessor;
+
+ @BeforeEach
+ public void setup() {
+ Kinesis2Component component = new Kinesis2Component(context);
+ component.start();
+ Kinesis2Configuration configuration = new Kinesis2Configuration();
+ configuration.setStreamName("stream");
+ configuration.setApplicationName("app");
+ // Provide a client so the endpoint does not build a real AWS client
(which needs a region) on start.
+ configuration.setAmazonKinesisClient(kinesisClient);
+ Kinesis2Endpoint endpoint = new
Kinesis2Endpoint("aws2-kinesis:stream", configuration, component);
+ endpoint.start();
+ KclKinesis2Consumer consumer = new KclKinesis2Consumer(endpoint,
processor);
+ recordProcessor = consumer.new CamelKinesisRecordProcessor(endpoint);
+
+ when(record.data()).thenReturn(ByteBuffer.wrap("hello".getBytes()));
+ when(record.partitionKey()).thenReturn("pk");
+ when(record.sequenceNumber()).thenReturn("1");
+ }
+
+ private ProcessRecordsInput input() {
+ return
ProcessRecordsInput.builder().records(List.of(record)).checkpointer(checkpointer).build();
+ }
+
+ @Test
+ public void checkpointsAfterSuccessfulBatch() throws Exception {
+ recordProcessor.processRecords(input());
+
+ verify(processor, times(1)).process(any(Exchange.class));
+ verify(checkpointer, times(1)).checkpoint();
+ }
+
+ @Test
+ public void doesNotCheckpointWhenProcessingFails() throws Exception {
+ doThrow(new RuntimeException("processing
failed")).when(processor).process(any(Exchange.class));
+
+ recordProcessor.processRecords(input());
+
+ // The failed batch must not be checkpointed, so KCL redelivers it
from the last checkpoint.
+ verify(checkpointer, never()).checkpoint();
+ }
+}
