CAMEL-24191: camel-aws2-kinesis - fix KCL consumer session credentials and checkpoint/error handling

components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java

@@ -92,19 +92,19 @@ 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())));
             } else if (ObjectHelper.isNotEmpty(configuration.getProfileCredentialsName())) {
                 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,19 +117,19 @@ 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())));
             } else if (ObjectHelper.isNotEmpty(configuration.getProfileCredentialsName())) {
                 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);
             }
         }
 

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();
+    }
+}