#1457-Introduce-kinesis-autoconfiguration

pom.xml

@@ -45,11 +45,11 @@
 		<module>spring-cloud-aws-sqs</module>
 		<module>spring-cloud-aws-dynamodb</module>
 		<module>spring-cloud-aws-kinesis</module>
-		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-kinesis</module>
 		<module>spring-cloud-aws-s3</module>
 		<module>spring-cloud-aws-testcontainers</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-dynamodb</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-kinesis</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-dynamodb</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-metrics</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-parameter-store</module>
@@ -61,6 +61,9 @@
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-sns</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-sqs</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-sqs</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-kinesis</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-kinesis-producer</module>
+		<module>spring-cloud-aws-starters/spring-cloud-aws-starter-kinesis-client-library</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.awssdk</groupId>
+			<artifactId>kinesis</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>software.amazon.kinesis</groupId>
+			<artifactId>amazon-kinesis-client</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>software.amazon.kinesis</groupId>
+			<artifactId>amazon-kinesis-producer</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-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.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/KinesisClientLibraryAutoConfiguration.java

@@ -0,0 +1,49 @@
+/*
+ * 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.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.cloudwatch.CloudWatchAsyncClient;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
+import software.amazon.kinesis.coordinator.Scheduler;
+import software.amazon.kinesis.processor.ShardRecordProcessorFactory;
+
+@AutoConfiguration
+@ConditionalOnClass({ KinesisAsyncClient.class, Scheduler.class })
+@EnableConfigurationProperties({ KinesisClientLibraryProperties.class })
+@AutoConfigureAfter({ CredentialsProviderAutoConfiguration.class, RegionProviderAutoConfiguration.class,
+		KinesisAutoConfiguration.class })
+@ConditionalOnProperty(value = "spring.cloud.aws.kinesis.client.library.enabled", havingValue = "true", matchIfMissing = true)
+public class KinesisClientLibraryAutoConfiguration {
+
+	@ConditionalOnMissingBean
+	@Bean
+	public Scheduler scheduler(ObjectProvider<DynamoDbAsyncClient> dynamoDbClient,
+			ObjectProvider<CloudWatchAsyncClient> cloudWatchClient, KinesisAsyncClient kinesisAsyncClient,
+			KinesisClientLibraryProperties properties, ShardRecordProcessorFactory processorFactory) {
+		return null;
+	}
+}

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

@@ -0,0 +1,45 @@
+/*
+ * 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.KinesisClientLibraryProperties.PREFIX;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = PREFIX)
+public class KinesisClientLibraryProperties {
+
+	public static final String PREFIX = "spring.cloud.aws.kinesis.client.library";
+
+	private String streamName;
+	private String applicationName;
+
+	public String getStreamName() {
+		return streamName;
+	}
+
+	public void setStreamName(String streamName) {
+		this.streamName = streamName;
+	}
+
+	public String getApplicationName() {
+		return applicationName;
+	}
+
+	public void setApplicationName(String applicationName) {
+		this.applicationName = applicationName;
+	}
+}

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

@@ -0,0 +1,100 @@
+/*
+ * 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.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.boot.context.properties.PropertyMapper;
+import org.springframework.context.annotation.Bean;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.regions.providers.AwsRegionProvider;
+import software.amazon.kinesis.producer.KinesisProducer;
+import software.amazon.kinesis.producer.KinesisProducerConfiguration;
+
+@AutoConfiguration
+@ConditionalOnClass({ KinesisProducer.class, KinesisProducerConfiguration.class })
+@EnableConfigurationProperties({ KinesisProducerProperties.class })
+@AutoConfigureAfter({ CredentialsProviderAutoConfiguration.class, RegionProviderAutoConfiguration.class })
+@ConditionalOnProperty(value = "spring.cloud.aws.kinesis.producer.enabled", havingValue = "true", matchIfMissing = true)
+public class KinesisProducerAutoConfiguration {
+
+	@ConditionalOnMissingBean
+	@Bean
+	public KinesisProducerConfiguration kinesisProducerConfiguration(KinesisProducerProperties prop,
+			AwsCredentialsProvider credentialsProvider, AwsRegionProvider awsRegionProvider,
+			ObjectProvider<AwsConnectionDetails> connectionDetails) {
+		PropertyMapper propertyMapper = PropertyMapper.get();
+		KinesisProducerConfiguration config = new KinesisProducerConfiguration();
+		propertyMapper.from(prop::getAggregationEnabled).whenNonNull().to(config::setAggregationEnabled);
+		propertyMapper.from(prop::getAggregationMaxCount).whenNonNull().to(config::setAggregationMaxCount);
+		propertyMapper.from(prop::getAggregationMaxSize).whenNonNull().to(config::setAggregationMaxSize);
+		propertyMapper.from(prop::getCloudwatchEndpoint).whenHasText().to(config::setCloudwatchEndpoint);
+		propertyMapper.from(prop::getCloudwatchPort).whenNonNull().to(config::setCloudwatchPort);
+		propertyMapper.from(prop::getCollectionMaxCount).whenNonNull().to(config::setCollectionMaxCount);
+		propertyMapper.from(prop::getCollectionMaxSize).whenNonNull().to(config::setCollectionMaxSize);
+		propertyMapper.from(prop::getConnectTimeout).whenNonNull().to(config::setConnectTimeout);
+		propertyMapper.from(prop::getCredentialsRefreshDelay).whenNonNull().to(config::setCredentialsRefreshDelay);
+		propertyMapper.from(prop::getEnableCoreDumps).whenNonNull().to(config::setEnableCoreDumps);
+		propertyMapper.from(prop::getFailIfThrottled).whenNonNull().to(config::setFailIfThrottled);
+		propertyMapper.from(prop::getLogLevel).whenHasText().to(config::setLogLevel);
+		propertyMapper.from(prop::getMaxConnections).whenNonNull().to(config::setMaxConnections);
+		propertyMapper.from(prop::getMetricsGranularity).whenHasText().to(config::setMetricsGranularity);
+		propertyMapper.from(prop::getMetricsLevel).whenHasText().to(config::setMetricsLevel);
+		propertyMapper.from(prop::getMetricsNamespace).whenHasText().to(config::setMetricsNamespace);
+		propertyMapper.from(prop::getMetricsUploadDelay).whenNonNull().to(config::setMetricsUploadDelay);
+		propertyMapper.from(prop::getMinConnections).whenNonNull().to(config::setMinConnections);
+		propertyMapper.from(prop::getNativeExecutable).whenNonNull().to(config::setNativeExecutable);
+		propertyMapper.from(prop::getRateLimit).whenNonNull().to(config::setRateLimit);
+		propertyMapper.from(prop::getRecordMaxBufferedTime).whenNonNull().to(config::setRecordMaxBufferedTime);
+		propertyMapper.from(prop::getRecordTtl).whenNonNull().to(config::setRecordTtl);
+		propertyMapper.from(prop::getRequestTimeout).whenNonNull().to(config::setRequestTimeout);
+		propertyMapper.from(prop::getTempDirectory).whenNonNull().to(config::setTempDirectory);
+		propertyMapper.from(prop::getVerifyCertificate).whenNonNull().to(config::setVerifyCertificate);
+		propertyMapper.from(prop.getProxyHost()).whenNonNull().to(config::setProxyHost);
+		propertyMapper.from(prop.getProxyPort()).whenNonNull().to(config::setProxyPort);
+		propertyMapper.from(prop.getProxyUserName()).whenHasText().to(config::setProxyUserName);
+		propertyMapper.from(prop.getProxyPassword()).whenHasText().to(config::setProxyPassword);
+		propertyMapper.from(prop.getStsEndpoint()).whenHasText().to(config::setStsEndpoint);
+		propertyMapper.from(prop.getStsPort()).whenNonNull().to(config::setStsPort);
+		propertyMapper.from(prop.getThreadingModel()).whenNonNull().to(config::setThreadingModel);
+		propertyMapper.from(prop.getThreadPoolSize()).whenNonNull().to(config::setThreadPoolSize);
+		propertyMapper.from(prop.getUserRecordTimeoutInMillis()).whenNonNull().to(config::setUserRecordTimeoutInMillis);
+
+		config.setCredentialsProvider(credentialsProvider);
+		config.setRegion(AwsClientBuilderConfigurer
+				.resolveRegion(prop, connectionDetails.getIfAvailable(), awsRegionProvider).toString());
+		connectionDetails.ifAvailable(cd -> {
+			config.setKinesisPort(cd.getEndpoint().getPort());
+			config.setKinesisEndpoint(cd.getEndpoint().getHost());
+		});
+		return config;
+	}
+
+	@ConditionalOnMissingBean
+	@Bean
+	public KinesisProducer kinesisProducer(KinesisProducerConfiguration kinesisProducerConfiguration) {
+		return new KinesisProducer(kinesisProducerConfiguration);
+	}
+}