GH-230: Expose kplBackPressureThreshold property

Fixes: #230

Author: artembilan

spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc

@@ -359,6 +359,11 @@ recordMetadataChannel::
 The bean name of a MessageChannel to which successful send results should be sent.
 Works only for async mode.
 
+kplBackPressureThreshold::
+Maximum records in flight for handling backpressure.
+No backpressure by default.
+When backpressure handling is enabled and number of records in flight exceeds the threshold, a `KplBackpressureException` would be thrown.
+
 [[kinesis-error-channels]]
 == Error Channels
 

spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2023 the original author or authors.
+ * Copyright 2017-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.
@@ -224,18 +224,19 @@ protected MessageHandler createProducerMessageHandler(ProducerDestination destin
 								? m.getHeaders().get(BinderHeaders.PARTITION_HEADER)
 								: m.getPayload().hashCode());
 		AbstractAwsMessageHandler<?> messageHandler;
+		KinesisProducerProperties kinesisProducerProperties = producerProperties.getExtension();
 		if (this.configurationProperties.isKplKclEnabled()) {
-			messageHandler = createKplMessageHandler(destination, partitionKeyExpression,
-					producerProperties.getExtension().isEmbedHeaders() && !producerProperties.isUseNativeEncoding());
+			messageHandler = createKplMessageHandler(destination, partitionKeyExpression, kinesisProducerProperties,
+					kinesisProducerProperties.isEmbedHeaders() && !producerProperties.isUseNativeEncoding());
 		}
 		else {
 			messageHandler = createKinesisMessageHandler(destination, partitionKeyExpression,
-					producerProperties.getExtension().isEmbedHeaders() && !producerProperties.isUseNativeEncoding());
+					kinesisProducerProperties.isEmbedHeaders() && !producerProperties.isUseNativeEncoding());
 		}
-		messageHandler.setAsync(!producerProperties.getExtension().isSync());
-		messageHandler.setSendTimeout(producerProperties.getExtension().getSendTimeout());
+		messageHandler.setAsync(!kinesisProducerProperties.isSync());
+		messageHandler.setSendTimeout(kinesisProducerProperties.getSendTimeout());
 		messageHandler.setBeanFactory(getBeanFactory());
-		String recordMetadataChannel = producerProperties.getExtension().getRecordMetadataChannel();
+		String recordMetadataChannel = kinesisProducerProperties.getRecordMetadataChannel();
 		if (StringUtils.hasText(recordMetadataChannel)) {
 			messageHandler.setOutputChannelName(recordMetadataChannel);
 		}
@@ -279,11 +280,13 @@ private AbstractAwsMessageHandler<?> createKinesisMessageHandler(ProducerDestina
 	}
 
 	private AbstractAwsMessageHandler<?> createKplMessageHandler(ProducerDestination destination,
-			FunctionExpression<Message<?>> partitionKeyExpression, boolean embedHeaders) {
+			FunctionExpression<Message<?>> partitionKeyExpression, KinesisProducerProperties kinesisProducerProperties,
+			boolean embedHeaders) {
 
 		KplMessageHandler messageHandler = new KplMessageHandler(new KinesisProducer(this.kinesisProducerConfiguration));
 		messageHandler.setStream(destination.getName());
 		messageHandler.setPartitionKeyExpression(partitionKeyExpression);
+		messageHandler.setBackPressureThreshold(kinesisProducerProperties.getKplBackPressureThreshold());
 		if (embedHeaders) {
 			messageHandler.setEmbeddedHeadersMapper(this.embeddedHeadersMapper);
 		}

spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisProducerProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2024 the original author or authors.
+ * Copyright 2017-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.
@@ -46,6 +46,14 @@ public class KinesisProducerProperties {
 	 */
 	private String recordMetadataChannel;
 
+	/**
+	 * Maximum records in flight for handling backpressure.
+	 * No backpressure by default.
+	 * When backpressure handling is enabled and number of records in flight exceeds the threshold, a
+	 * 'KplBackpressureException' would be thrown.
+	 */
+	private long kplBackPressureThreshold;
+
 	public void setSync(boolean sync) {
 		this.sync = sync;
 	}
@@ -78,4 +86,12 @@ public void setRecordMetadataChannel(String recordMetadataChannel) {
 		this.recordMetadataChannel = recordMetadataChannel;
 	}
 
+	public long getKplBackPressureThreshold() {
+		return this.kplBackPressureThreshold;
+	}
+
+	public void setKplBackPressureThreshold(long kplBackPressureThreshold) {
+		this.kplBackPressureThreshold = kplBackPressureThreshold;
+	}
+
 }