Add Kinesis Producer Library with Spring Integration (#1493)

* Add `KplMessageHandler` - a `MessageHandler` implementation to produce into Kinesis via KPL
* Add dedicated to Spring Integration with KPL starter - `spring-cloud-aws-starter-integration-kinesis-producer`
* Document the `KplMessageHandler`

Author: artembilan

docs/src/main/asciidoc/kinesis.adoc

@@ -114,8 +114,42 @@ public PollableChannel errorChannel() {
 }
 ----
 
+The `KplMessageHandler` is an `AbstractMessageHandler` to perform put record to the Kinesis stream using https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html[Kinesis Producer Library (KPL)].
+The configuration and behavior are similar to the `KinesisMessageHandler` described above, with the difference that it supports only a single `UserRecord` production according to KPL API.
+The request message payload could be as a `UserRecord`.
+Otherwise, such an instance is built against request messages and respective `KplMessageHandler` options, include https://docs.aws.amazon.com/streams/latest/dev/kpl-with-schemaregistry.html[AWS Glue Schema] for the record serialization.
+
+Due to asynchronous behavior and buffer of the `KinesisProducer`, a `KplBackpressureException` could be thrown from the  `KplMessageHandler` when `backPressureThreshold` as the number of outstanding records is provided.
+
+The configuration of the `KplMessageHandler` is straightforward:
+
+[source,java]
+----
+@Bean
+RequestHandlerRetryAdvice retryAdvice() {
+    RequestHandlerRetryAdvice requestHandlerRetryAdvice = new RequestHandlerRetryAdvice();
+    requestHandlerRetryAdvice.setRetryTemplate(RetryTemplate.builder()
+        .retryOn(KplBackpressureException.class)
+        .exponentialBackoff(100, 2.0, 1000)
+        .maxAttempts(3)
+        .build());
+    return requestHandlerRetryAdvice;
+}
+
+@Bean
+@ServiceActivator(inputChannel = "kinesisSendChannel", adviceChain = "retryAdvice")
+MessageHandler kplMessageHandler(KinesisProducer kinesisProducer, Schema schema) {
+    KplMessageHandler kplMessageHandler = new KplMessageHandler(kinesisProducer);
+    kplMessageHandler.setAsync(true);
+    kplMessageHandler.setStream("someStream");
+	kplMessageHandler.setBackPressureThreshold(2);
+    kplMessageHandler.setGlueSchema(schema);
+    return kplMessageHandler;
+}
+----
 
 === Spring Integration Starters
 
 The Spring Integration dependency in the `spring-cloud-aws-kinesis` module is `optional` to avoid unnecessary artifacts on classpath when Spring Integration is not used.
 For convenience, a dedicated `spring-cloud-aws-starter-integration-kinesis` is provided managing all the required dependencies for Spring Integration support with a classical Amazon Kinesis client.
+The `spring-cloud-aws-starter-integration-kinesis-producer` artifact is dedicated for dependencies related to the Kinesis Producer Library.

pom.xml

@@ -46,6 +46,7 @@
 		<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-starters/spring-cloud-aws-starter-integration-kinesis-producer</module>
 		<module>spring-cloud-aws-s3</module>
 		<module>spring-cloud-aws-testcontainers</module>
 		<module>spring-cloud-aws-starters/spring-cloud-aws-starter</module>

spring-cloud-aws-dependencies/pom.xml

@@ -123,6 +123,11 @@
 				<artifactId>spring-cloud-aws-starter-integration-kinesis</artifactId>
 				<version>${project.version}</version>
 			</dependency>
+			<dependency>
+				<groupId>io.awspring.cloud</groupId>
+				<artifactId>spring-cloud-aws-starter-integration-kinesis-producer</artifactId>
+				<version>${project.version}</version>
+			</dependency>
 
 			<dependency>
 				<groupId>io.awspring.cloud</groupId>

spring-cloud-aws-kinesis/src/main/java/io/awspring/cloud/kinesis/integration/KplBackpressureException.java

@@ -0,0 +1,50 @@
+/*
+ * 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.kinesis.integration;
+
+import java.io.Serial;
+import software.amazon.kinesis.producer.UserRecord;
+
+/**
+ * An exception triggered from the {@link KplMessageHandler} while sending records to Kinesis when maximum number of
+ * records in flight exceeds the backpressure threshold.
+ *
+ * @author Siddharth Jain
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class KplBackpressureException extends RuntimeException {
+
+	@Serial
+	private static final long serialVersionUID = 1L;
+
+	private final transient UserRecord userRecord;
+
+	public KplBackpressureException(String message, UserRecord userRecord) {
+		super(message);
+		this.userRecord = userRecord;
+	}
+
+	/**
+	 * Get the {@link UserRecord} when this exception has been thrown.
+	 * @return the {@link UserRecord} when this exception has been thrown.
+	 */
+	public UserRecord getUserRecord() {
+		return this.userRecord;
+	}
+
+}