Add KclMessageDrivenChannelAdapter.leaseTableName property

Author: artembilan

src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java

@@ -23,8 +23,6 @@
 import java.util.UUID;
 import java.util.function.Consumer;
 
-import javax.annotation.Nullable;
-
 import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializer;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -87,6 +85,7 @@
 import org.springframework.integration.support.ErrorMessageUtils;
 import org.springframework.integration.support.management.IntegrationManagedResource;
 import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
@@ -123,6 +122,9 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport
 
 	private String consumerGroup = "SpringIntegration";
 
+	@Nullable
+	private String leaseTableName;
+
 	private InboundMessageMapper<byte[]> embeddedHeadersMapper;
 
 	private ConfigsBuilder config;
@@ -211,6 +213,16 @@ public String getConsumerGroup() {
 		return this.consumerGroup;
 	}
 
+	/**
+	 * Set a name of the DynamoDB table name for leases.
+	 * Defaults to {@link #consumerGroup}.
+	 * @param leaseTableName the DynamoDB table name for leases.
+	 * @since 3.0.8
+	 */
+	public void setLeaseTableName(String leaseTableName) {
+		this.leaseTableName = leaseTableName;
+	}
+
 	/**
 	 * Specify an {@link InboundMessageMapper} to extract message headers embedded into
 	 * the record data.
@@ -372,6 +384,10 @@ protected void onInit() {
 						this.cloudWatchClient,
 						this.workerId,
 						this.recordProcessorFactory);
+
+		if (this.leaseTableName != null) {
+			this.config.tableName(this.leaseTableName);
+		}
 	}
 
 	private StreamTracker buildStreamTracker() {

src/test/java/org/springframework/integration/aws/kinesis/KclMessageDrivenChannelAdapterTests.java

@@ -121,6 +121,9 @@ void kclChannelAdapterReceivesRecords() {
 
 		// Because FanOut is false, there would be no Stream Consumers.
 		assertThat(streamConsumers).hasSize(0);
+
+		List<String> tableNames = DYNAMO_DB.listTables().join().tableNames();
+		assertThat(tableNames).containsOnly("test_table");
 	}
 
 	@Test
@@ -172,6 +175,7 @@ public KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter() {
 					InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON));
 			adapter.setConverter(String::new);
 			adapter.setConsumerGroup("single_stream_group");
+			adapter.setLeaseTableName("test_table");
 			adapter.setFanOut(false);
 			adapter.setMetricsLevel(MetricsLevel.NONE);
 			adapter.setLeaseManagementConfigCustomizer(leaseManagementConfig ->