#1457-Introduce-kinesis-autoconfiguration

pom.xml

@@ -56,6 +56,7 @@
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-ses</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-sns</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-sqs</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-kinesis</module>
 		<module>spring-cloud-aws-samples</module>
 		<module>spring-cloud-aws-test</module>
 		<module>spring-cloud-aws-modulith</module>

spring-cloud-aws-autoconfigure/pom.xml

@@ -181,6 +181,21 @@
 			<artifactId>kms</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>software.amazon.kinesis</groupId>
+			<artifactId>amazon-kinesis-client</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>com.amazonaws</groupId>
+			<artifactId>amazon-kinesis-producer</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>software.amazon.awssdk</groupId>
+			<artifactId>kinesis</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>software.amazon.encryption.s3</groupId>
 			<artifactId>amazon-s3-encryption-client-java</artifactId>

spring-cloud-aws-autoconfigure/src/main/java/io/awspring/cloud/autoconfigure/kinesis/KinesisAsyncClientCustomizer.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 io.awspring.cloud.autoconfigure.kinesis;
+
+import io.awspring.cloud.autoconfigure.AwsClientCustomizer;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClientBuilder;
+
+/**
+ * Callback interface that can be used to customize a {@link KinesisAsyncClientBuilder}.
+ *
+ * @author Matej Nedic
+ * @since 4.0.0
+ */
+@FunctionalInterface
+public interface KinesisAsyncClientCustomizer extends AwsClientCustomizer<KinesisAsyncClientBuilder> {
+}

spring-cloud-aws-autoconfigure/src/main/java/io/awspring/cloud/autoconfigure/kinesis/KinesisAutoConfiguration.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2013-2025 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 io.awspring.cloud.autoconfigure.kinesis;
+
+import io.awspring.cloud.autoconfigure.AwsAsyncClientCustomizer;
+import io.awspring.cloud.autoconfigure.core.AwsClientBuilderConfigurer;
+import io.awspring.cloud.autoconfigure.core.AwsConnectionDetails;
+import io.awspring.cloud.autoconfigure.core.CredentialsProviderAutoConfiguration;
+import io.awspring.cloud.autoconfigure.core.RegionProviderAutoConfiguration;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+
+@AutoConfiguration
+@ConditionalOnClass({ KinesisAsyncClient.class })
+@EnableConfigurationProperties({ KinesisProperties.class })
+@AutoConfigureAfter({ CredentialsProviderAutoConfiguration.class, RegionProviderAutoConfiguration.class })
+@ConditionalOnProperty(value = "spring.cloud.aws.kinesis.enabled", havingValue = "true", matchIfMissing = true)
+public class KinesisAutoConfiguration {
+
+	@ConditionalOnMissingBean
+	@Bean
+	public KinesisAsyncClient kinesisAsyncClient(KinesisProperties properties,
+			AwsClientBuilderConfigurer awsClientBuilderConfigurer,
+			ObjectProvider<AwsConnectionDetails> connectionDetails,
+			ObjectProvider<KinesisAsyncClientCustomizer> kinesisAsyncClientCustomizer,
+			ObjectProvider<AwsAsyncClientCustomizer> awsSyncClientCustomizers) {
+		return awsClientBuilderConfigurer
+				.configureAsyncClient(KinesisAsyncClient.builder(), properties, connectionDetails.getIfAvailable(),
+						kinesisAsyncClientCustomizer.orderedStream(), awsSyncClientCustomizers.orderedStream())
+				.build();
+	}
+}

spring-cloud-aws-autoconfigure/src/main/java/io/awspring/cloud/autoconfigure/kinesis/KinesisProperties.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2013-2025 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 io.awspring.cloud.autoconfigure.kinesis;
+
+import static io.awspring.cloud.autoconfigure.kinesis.KinesisProperties.PREFIX;
+
+import io.awspring.cloud.autoconfigure.AwsClientProperties;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Properties related to KinesisClient
+ *
+ * @author Matej Nedic
+ * @since 4.0.0
+ */
+@ConfigurationProperties(prefix = PREFIX)
+public class KinesisProperties extends AwsClientProperties {
+	/**
+	 * The prefix used for AWS Kinesis configuration.
+	 */
+	public static final String PREFIX = "spring.cloud.aws.kinesis";
+}

spring-cloud-aws-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -15,3 +15,4 @@ io.awspring.cloud.autoconfigure.config.secretsmanager.SecretsManagerAutoConfigur
 io.awspring.cloud.autoconfigure.config.parameterstore.ParameterStoreReloadAutoConfiguration
 io.awspring.cloud.autoconfigure.config.parameterstore.ParameterStoreAutoConfiguration
 io.awspring.cloud.autoconfigure.config.s3.S3ReloadAutoConfiguration
+io.awspring.cloud.autoconfigure.kinesis.KinesisAutoConfiguration

spring-cloud-aws-autoconfigure/src/test/java/io/awspring/cloud/autoconfigure/kinesis/KinesisAutoConfigurationTest.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2013-2025 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 io.awspring.cloud.autoconfigure.kinesis;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.awspring.cloud.autoconfigure.ConfiguredAwsClient;
+import io.awspring.cloud.autoconfigure.core.AwsAutoConfiguration;
+import io.awspring.cloud.autoconfigure.core.CredentialsProviderAutoConfiguration;
+import io.awspring.cloud.autoconfigure.core.RegionProviderAutoConfiguration;
+import java.net.URI;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+
+/**
+ * Tests for {@link KinesisAutoConfiguration}.
+ *
+ * @author Matej Nedic
+ */
+class KinesisAutoConfigurationTest {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withPropertyValues("spring.cloud.aws.region.static:eu-west-1",
+					"spring.cloud.aws.credentials.access-key:noop", "spring.cloud.aws.credentials.secret-key:noop")
+			.withConfiguration(AutoConfigurations.of(AwsAutoConfiguration.class, RegionProviderAutoConfiguration.class,
+					CredentialsProviderAutoConfiguration.class, KinesisAutoConfiguration.class));
+
+	@Test
+	void disableKinesisIntegration() {
+		this.contextRunner.withPropertyValues("spring.cloud.aws.kinesis.enabled:false").run(context -> {
+			assertThat(context).doesNotHaveBean(KinesisAsyncClient.class);
+		});
+	}
+
+	@Test
+	void withCustomEndpoint() {
+		this.contextRunner.withPropertyValues("spring.cloud.aws.kinesis.endpoint:http://localhost:8090")
+				.run(context -> {
+					ConfiguredAwsClient client = new ConfiguredAwsClient(context.getBean(KinesisAsyncClient.class));
+					assertThat(client.getEndpoint()).isEqualTo(URI.create("http://localhost:8090"));
+					assertThat(client.isEndpointOverridden()).isTrue();
+				});
+	}
+}

spring-cloud-aws-autoconfigure/src/test/java/io/awspring/cloud/autoconfigure/kinesis/KinesisClientCustomizerTests.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2013-2025 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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 io.awspring.cloud.autoconfigure.kinesis;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.awspring.cloud.autoconfigure.ConfiguredAwsClient;
+import io.awspring.cloud.autoconfigure.core.AwsAutoConfiguration;
+import io.awspring.cloud.autoconfigure.core.CredentialsProviderAutoConfiguration;
+import io.awspring.cloud.autoconfigure.core.RegionProviderAutoConfiguration;
+import java.time.Duration;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+
+public class KinesisClientCustomizerTests {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withPropertyValues("spring.cloud.aws.region.static:eu-west-1",
+					"spring.cloud.aws.credentials.access-key:noop", "spring.cloud.aws.credentials.secret-key:noop")
+			.withConfiguration(AutoConfigurations.of(AwsAutoConfiguration.class, RegionProviderAutoConfiguration.class,
+					CredentialsProviderAutoConfiguration.class, KinesisAutoConfiguration.class));
+
+	@Test
+	void customKinesisClientCustomizer() {
+		contextRunner.withUserConfiguration(KinesisClientCustomizerTests.CustomizerConfig.class).run(context -> {
+			ConfiguredAwsClient kinesisAsyncClient = new ConfiguredAwsClient(context.getBean(KinesisAsyncClient.class));
+			assertThat(kinesisAsyncClient.getApiCallTimeout()).describedAs("sets property from first customizer")
+					.isEqualTo(Duration.ofMillis(2001));
+			assertThat(kinesisAsyncClient.getApiCallAttemptTimeout())
+					.describedAs("sets property from second customizer").isEqualTo(Duration.ofMillis(2002));
+		});
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	static class CustomizerConfig {
+
+		@Bean
+		KinesisAsyncClientCustomizer kinesisClientCustomizer() {
+			return builder -> {
+				builder.overrideConfiguration(builder.overrideConfiguration().copy(c -> {
+					c.apiCallTimeout(Duration.ofMillis(2001));
+				}));
+			};
+		}
+
+		@Bean
+		KinesisAsyncClientCustomizer kinesisClientCustomizer2() {
+			return builder -> {
+				builder.overrideConfiguration(builder.overrideConfiguration().copy(c -> {
+					c.apiCallAttemptTimeout(Duration.ofMillis(2002));
+				}));
+			};
+		}
+	}
+}

spring-cloud-aws-dependencies/pom.xml

@@ -33,6 +33,8 @@
 		<eclipse.jakarta.mail.version>2.0.3</eclipse.jakarta.mail.version>
 		<spring-modulith.version>2.0.0-M2</spring-modulith.version>
 		<amazon.s3.accessgrants>2.3.0</amazon.s3.accessgrants>
+		<amazon.kinesis.client>2.7.0</amazon.kinesis.client>
+		<amazon.kinesis.producer>0.15.12</amazon.kinesis.producer>
 	</properties>
 
 	<dependencyManagement>
@@ -68,13 +70,27 @@
 				<optional>true</optional>
 			</dependency>
 
+			<dependency>
+				<groupId>software.amazon.kinesis</groupId>
+				<artifactId>amazon-kinesis-client</artifactId>
+				<version>${amazon.kinesis.client}</version>
+				<optional>true</optional>
+			</dependency>
+
 			<dependency>
 				<groupId>software.amazon.encryption.s3</groupId>
 				<artifactId>amazon-s3-encryption-client-java</artifactId>
 				<version>${amazon.encryption.s3.version}</version>
 				<optional>true</optional>
 			</dependency>
 
+			<dependency>
+				<groupId>com.amazonaws</groupId>
+				<artifactId>amazon-kinesis-producer</artifactId>
+				<version>${amazon.kinesis.producer}</version>
+				<optional>true</optional>
+			</dependency>
+
 			<dependency>
 				<groupId>software.amazon.s3.accessgrants</groupId>
 				<artifactId>aws-s3-accessgrants-java-plugin</artifactId>

spring-cloud-aws-starters/spring-cloud-aws-starter-kinesis/pom.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<parent>
+		<artifactId>spring-cloud-aws</artifactId>
+		<groupId>io.awspring.cloud</groupId>
+		<version>4.0.0-SNAPSHOT</version>
+		<relativePath>../../pom.xml</relativePath>
+	</parent>
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-aws-starter-kinesis</artifactId>
+	<name>Spring Cloud AWS Kinesis Starter</name>
+	<description>Spring Cloud AWS Kinesis Starter</description>
+
+	<dependencies>
+		<dependency>
+			<groupId>io.awspring.cloud</groupId>
+			<artifactId>spring-cloud-aws-starter</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>software.amazon.awssdk</groupId>
+			<artifactId>kinesis</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>software.amazon.kinesis</groupId>
+			<artifactId>amazon-kinesis-client</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>com.amazonaws</groupId>
+			<artifactId>amazon-kinesis-producer</artifactId>
+		</dependency>
+	</dependencies>
+</project>