From 008f57b95cb6720c7ae8832792c6b23d81afa420 Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Fri, 17 Jul 2026 10:56:42 +0200 Subject: [PATCH 1/2] CAMEL-24191: camel-aws2-kinesis - fix KCL consumer session credentials and checkpoint/error handling Two defects in the KCL-based Kinesis consumer: - In both the DynamoDB and CloudWatch async-client builders the credential branches tested accessKey+secretKey before accessKey+secretKey+sessionToken, so the session-token branch was unreachable and those clients never used STS temporary credentials. The session-credentials branch is now checked first. - processRecords wrapped processing in catch(Throwable) that logged without the exception and swallowed it, so the scheduler later checkpointed past records that failed to process (data loss), and it never checkpointed after a successful batch (a crash reprocessed the whole lease). It now checkpoints only after the batch is processed successfully, and on failure routes the exception to the consumer ExceptionHandler without checkpointing, so KCL redelivers the batch from the last checkpoint. Adds KclKinesis2ConsumerRecordProcessingTest asserting the checkpoint is taken after a successful batch and skipped when processing fails. Related to CAMEL-24156. Co-Authored-By: Claude Fable 5 Signed-off-by: Andrea Cosentino --- .../aws2/kinesis/KclKinesis2Consumer.java | 47 ++++----- ...lKinesis2ConsumerRecordProcessingTest.java | 99 +++++++++++++++++++ 2 files changed, 124 insertions(+), 22 deletions(-) create mode 100644 components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KclKinesis2ConsumerRecordProcessingTest.java 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 8e1a25bda35b5..d4b04a2b7d0bb 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 @@ protected void doStart() throws Exception { 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 @@ protected void doStart() throws Exception { 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 @@ protected void doStart() throws Exception { 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 @@ protected void doStart() throws Exception { 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 void initialize(InitializationInput initializationInput) { 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 0000000000000..e81daaed14797 --- /dev/null +++ b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KclKinesis2ConsumerRecordProcessingTest.java @@ -0,0 +1,99 @@ +/* + * 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.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; + + 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"); + 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(); + } +} From c217bb45f909a715ee61f36288c8c899dbfa57fe Mon Sep 17 00:00:00 2001 From: Andrea Cosentino Date: Fri, 17 Jul 2026 11:37:01 +0200 Subject: [PATCH 2/2] CAMEL-24191: set a mock Kinesis client in the KCL record-processing test The test started the endpoint, which built a real Kinesis client and failed in CI with "Unable to load region" (no AWS_REGION configured). Provide a mock KinesisClient on the configuration so the endpoint does not build a real client on start, matching the sibling consumer tests. Co-Authored-By: Claude Fable 5 Signed-off-by: Andrea Cosentino --- .../kinesis/KclKinesis2ConsumerRecordProcessingTest.java | 5 +++++ 1 file changed, 5 insertions(+) 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 index e81daaed14797..5ec14f933b0f8 100644 --- 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 @@ -29,6 +29,7 @@ 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; @@ -54,6 +55,8 @@ public class KclKinesis2ConsumerRecordProcessingTest { private RecordProcessorCheckpointer checkpointer; @Mock private KinesisClientRecord record; + @Mock + private KinesisClient kinesisClient; private final DefaultCamelContext context = new DefaultCamelContext(); private KclKinesis2Consumer.CamelKinesisRecordProcessor recordProcessor; @@ -65,6 +68,8 @@ public void setup() { 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);