GH-224: More customization for KCL

Fixes: #224 Issue link: #224
spring-cloud · Sep 24, 2024 · ffc0f2e · ffc0f2e
1 parent 086199a
commit ffc0f2e
Show file tree

Hide file tree

Showing 5 changed files with 52 additions and 6 deletions.
diff --git a/pom.xml b/pom.xml
@@ -17,11 +17,11 @@
 	<properties>
 		<java.version>17</java.version>
 		<spring-cloud-stream.version>4.0.5</spring-cloud-stream.version>
-		<spring-cloud-aws.version>3.0.4</spring-cloud-aws.version>
-		<spring-integration-aws.version>3.0.6</spring-integration-aws.version>
-		<amazon-kinesis-client.version>2.5.7</amazon-kinesis-client.version>
-		<amazon-kinesis-producer.version>0.15.10</amazon-kinesis-producer.version>
-		<testcontainers.version>1.19.7</testcontainers.version>
+		<spring-cloud-aws.version>3.0.5</spring-cloud-aws.version>
+		<spring-integration-aws.version>3.0.8-SNAPSHOT</spring-integration-aws.version>
+		<amazon-kinesis-client.version>2.5.8</amazon-kinesis-client.version>
+		<amazon-kinesis-producer.version>0.15.11</amazon-kinesis-producer.version>
+		<testcontainers.version>1.20.1</testcontainers.version>
 	</properties>
 
 	<modules>

diff --git a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc
@@ -283,6 +283,36 @@ Whether to extract headers and payload from Kinesis record data.
 +
 Default: `false`
 
+emptyRecordList::
+Whether to emit an empty list of records into a consumer.
+Works only in `listenerMode.batch`.
++
+Default: `false`
+
+Starting with version `4.0.4` (basically since `spring-integration-aws-3.0.8`), the `KclMessageDrivenChannelAdapter` can be customized programmatically for the `ConfigsBuilder` parts.
+For example, to set a custom value for the `LeaseManagementConfig.maxLeasesForWorker` property, the `ConsumerEndpointCustomizer<KclMessageDrivenChannelAdapter>` bean has to be provided:
+
+[source,java]
+----
+@Bean
+ConsumerEndpointCustomizer<KclMessageDrivenChannelAdapter> consumerEndpointCustomizer() {
+    return (endpoint, destinationName, group) ->
+                 endpoint.setLeaseManagementConfigCustomizer(leaseManagementConfig ->
+                         leaseManagementConfig.maxLeasesForWorker(10));
+}
+----
+
+Other similar setters on the `KclMessageDrivenChannelAdapter` are:
+
+[source,java]
+----
+setCoordinatorConfigCustomizer(Consumer<CoordinatorConfig> coordinatorConfigCustomizer);
+
+setLifecycleConfigCustomizer(Consumer<LifecycleConfig> lifecycleConfigCustomizer);
+
+setMetricsConfigCustomizer(Consumer<MetricsConfig> metricsConfigCustomizer);
+----
+
 === Kinesis Producer Properties
 
 The following properties are available for Kinesis producers only and must be prefixed with `spring.cloud.stream.kinesis.bindings.<bindingTarget>.producer.`.

diff --git a/...ain/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java b/...ain/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java
@@ -413,6 +413,7 @@ private MessageProducerSupport createKclConsumerEndpoint(ConsumerDestination des
 		adapter.setListenerMode(kinesisConsumerProperties.getListenerMode());
 		adapter.setFanOut(kinesisConsumerProperties.isFanOut());
 		adapter.setMetricsLevel(kinesisConsumerProperties.getMetricsLevel());
+		adapter.setEmptyRecordList(kinesisConsumerProperties.isEmptyRecordList());
 		if (properties.getExtension().isEmbedHeaders()) {
 			adapter.setEmbeddedHeadersMapper(this.embeddedHeadersMapper);
 		}

diff --git a/...org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java b/...org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2023 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -67,6 +67,11 @@ public class KinesisConsumerProperties {
 	 */
 	private boolean fanOut = true;
 
+	/**
+	 * The KCL emptyRecordList option for batch listener mode.
+	 */
+	private boolean emptyRecordList = false;
+
 	private boolean embedHeaders;
 
 	/**
@@ -178,4 +183,12 @@ public void setMetricsLevel(MetricsLevel metricsLevel) {
 		this.metricsLevel = metricsLevel;
 	}
 
+	public boolean isEmptyRecordList() {
+		return this.emptyRecordList;
+	}
+
+	public void setEmptyRecordList(boolean emptyRecordList) {
+		this.emptyRecordList = emptyRecordList;
+	}
+
 }
diff --git a/...st/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderFunctionalTests.java b/...st/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderFunctionalTests.java
@@ -38,8 +38,10 @@
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binding.BindingService;
+import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChannelAdapter;
 import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter;
 import org.springframework.integration.aws.inbound.kinesis.KinesisShardOffset;
 import org.springframework.integration.aws.support.AwsHeaders;