Re-enable KinesisBinderFunctionalTests

Looks like the batch consumer for `Message<List<Message<?>>>` cannot be deduced yet.

* Use plain `Consumer<Message<?>>` for the `Message<List<Message<?>>>` from the Kinesis binder
* Fix typos in docs
* Optimize some code flows

Author: artembilan

docs/src/main/asciidoc/kinesis.adoc

@@ -181,13 +181,13 @@ NOTE: Unlike `KinesisMessageDrivenChannelAdapter`, the `KclMessageDrivenChannelA
 
 ==== DynamoDB Streams Support
 
-The `KinesisMessageDrivenChannelAdapter` and `KclMessageDrivenChannelAdapter` provides support for consuming CDC events from the https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html[DynamoDB Streams].
+The `KinesisMessageDrivenChannelAdapter` and `KclMessageDrivenChannelAdapter` provide support for consuming CDC events from the https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html[DynamoDB Streams].
 The `com.amazonaws:dynamodb-streams-kinesis-adapter` dependency must be present on classpath.
 The KCL provides native support for the mentioned adapter, and only instance of `DynamoDbStreamsClient` has to be injected into the `KclMessageDrivenChannelAdapter` to switch consuming logic from the Kinesis stream to DynamoDB stream.
 For consuming via `KinesisMessageDrivenChannelAdapter`, the `SpringDynamoDBStreamsAdapterClient` has to be injected instead of regular `KinesisAsyncClient`.
 The `SpringDynamoDBStreamsAdapterClient` is an extension of the `AmazonDynamoDBStreamsAdapterClient` with overridden `listShards()` and `getRecords()` operations to mimic `KinesisAsyncClient` API called from the `KinesisMessageDrivenChannelAdapter` logic.
 Both channel adapters require a Stream ARN for DynamoDB Stream on the table.
-Using AWS SDK API, such a value is available as a result of the `DescribeTableResponse.table().latestStreamArn()` in the answer to `DynamoDbAsyncClient.describeTable()` request.
+Using AWS SDK API, such a Stream ARN value is available as a result of the `DescribeTableResponse.table().latestStreamArn()` in the answer to `DynamoDbAsyncClient.describeTable()` request.
 
 ==== Spring Integration Starters
 

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/KinesisBinderHealthIndicator.java

@@ -27,7 +27,7 @@
 /**
  * @author Artem Bilan
  *
- * @since 2.0
+ * @since 4.0
  */
 public class KinesisBinderHealthIndicator implements HealthIndicator {
 

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/KinesisMessageChannelBinder.java

@@ -406,7 +406,7 @@ private MessageProducerSupport createKclConsumerEndpoint(ConsumerDestination des
 		adapter.setPollingMaxRecords(kinesisConsumerProperties.getPollingMaxRecords());
 		adapter.setPollingIdleTime(kinesisConsumerProperties.getPollingIdleTime());
 		adapter.setGracefulShutdownTimeout(kinesisConsumerProperties.getGracefulShutdownTimeout());
-		if (properties.getExtension().isEmbedHeaders() && !properties.isUseNativeDecoding()) {
+		if (kinesisConsumerProperties.isEmbedHeaders() && !properties.isUseNativeDecoding()) {
 			adapter.setEmbeddedHeadersMapper(this.embeddedHeadersMapper);
 		}
 
@@ -509,7 +509,7 @@ else if (shardId != null) {
 		adapter.setStreamInitialSequence(streamInitialSequence);
 
 		adapter.setListenerMode(kinesisConsumerProperties.getListenerMode());
-		if (properties.getExtension().isEmbedHeaders() && !properties.isUseNativeDecoding()) {
+		if (kinesisConsumerProperties.isEmbedHeaders() && !properties.isUseNativeDecoding()) {
 			adapter.setEmbeddedHeadersMapper(this.embeddedHeadersMapper);
 		}
 
@@ -559,8 +559,9 @@ private static String[] headersToMap(KinesisBinderConfigurationProperties config
 			headers.add("X-B3*");
 			headers.add("b3");
 		}
-		if (!ObjectUtils.isEmpty(configurationProperties.getHeaders())) {
-			Collections.addAll(headers, configurationProperties.getHeaders());
+		String[] additionalHeaders = configurationProperties.getHeaders();
+		if (!ObjectUtils.isEmpty(additionalHeaders)) {
+			Collections.addAll(headers, additionalHeaders);
 		}
 		return headers.toArray(new String[0]);
 	}

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/config/KinesisBinderConfiguration.java

@@ -160,6 +160,7 @@ public DynamoDbLockRepository dynamoDbLockRepository(@Autowired(required = false
 	@ConditionalOnProperty(name = "spring.cloud.stream.kinesis.binder.kpl-kcl-enabled", havingValue = "false", matchIfMissing = true)
 	public LockRegistry<?> dynamoDBLockRegistry(
 			@Autowired(required = false) DynamoDbLockRepository dynamoDbLockRepository) {
+
 		if (dynamoDbLockRepository != null) {
 			KinesisBinderConfigurationProperties.Locks locks = this.configurationProperties.getLocks();
 			DynamoDbLockRegistry dynamoDbLockRegistry = new DynamoDbLockRegistry(dynamoDbLockRepository);
@@ -185,8 +186,9 @@ public ConcurrentMetadataStore kinesisCheckpointStore(@Autowired(required = fals
 			kinesisCheckpointStore.setWriteCapacity(checkpoint.getWriteCapacity());
 			kinesisCheckpointStore.setCreateTableDelay(checkpoint.getCreateDelay());
 			kinesisCheckpointStore.setCreateTableRetries(checkpoint.getCreateRetries());
-			if (checkpoint.getTimeToLive() != null) {
-				kinesisCheckpointStore.setTimeToLive(checkpoint.getTimeToLive());
+			Integer timeToLive = checkpoint.getTimeToLive();
+			if (timeToLive != null) {
+				kinesisCheckpointStore.setTimeToLive(timeToLive);
 			}
 			return kinesisCheckpointStore;
 		}
@@ -200,6 +202,7 @@ public ConcurrentMetadataStore kinesisCheckpointStore(@Autowired(required = fals
 	@ConditionalOnProperty(name = "spring.cloud.stream.kinesis.binder.kpl-kcl-enabled")
 	public CloudWatchAsyncClient cloudWatch(CloudWatchProperties properties,
 			ObjectProvider<AwsClientCustomizer<CloudWatchAsyncClientBuilder>> configurer) {
+
 		if (this.hasInputs) {
 			return awsClientBuilderConfigurer.configureAsyncClient(CloudWatchAsyncClient.builder(), properties, null,
 					Stream.of(configurer.getIfAvailable()), null).build();
@@ -223,6 +226,7 @@ public KinesisProducerConfiguration kinesisProducerConfiguration() {
 	@ConditionalOnMissingBean
 	public DynamoDbStreamsClient dynamoDBStreams(DynamoDbStreamsProperties properties,
 			ObjectProvider<AwsClientCustomizer<DynamoDbStreamsClientBuilder>> configurer) {
+
 		if (this.hasInputs) {
 			return awsClientBuilderConfigurer
 					.configureAsyncClient(DynamoDbStreamsClient.builder(), properties, null, configurer.stream(), null)

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/properties/KinesisExtendedBindingProperties.java

@@ -57,8 +57,9 @@ public KinesisConsumerProperties getExtendedConsumerProperties(String channelNam
 
 	@Override
 	public KinesisProducerProperties getExtendedProducerProperties(String channelName) {
-		if (this.bindings.containsKey(channelName) && this.bindings.get(channelName).getProducer() != null) {
-			return this.bindings.get(channelName).getProducer();
+		KinesisBindingProperties kinesisBindingProperties = this.bindings.get(channelName);
+		if (kinesisBindingProperties != null && kinesisBindingProperties.getProducer() != null) {
+			return kinesisBindingProperties.getProducer();
 		}
 		else {
 			return new KinesisProducerProperties();

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/provisioning/KinesisStreamProvisioner.java

@@ -78,8 +78,7 @@ public ProducerDestination provisionProducerDestination(String name,
 
 		logger.info(() -> "Using Kinesis stream for outbound: " + name);
 
-		KinesisProducerProperties kinesisProducerProperties = properties.getExtension();
-		if (kinesisProducerProperties.isEmbedHeaders()) {
+		if (properties.getExtension().isEmbedHeaders()) {
 			properties.setHeaderMode(HeaderMode.none);
 		}
 

spring-cloud-aws-kinesis-stream-binder/src/test/java/io/awspring/cloud/kinesis/stream/binder/KinesisBinderFunctionalTests.java

@@ -30,7 +30,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import org.assertj.core.api.Condition;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -58,16 +57,15 @@
  * @since 4.0
  */
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
-		"spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.multiplex=true",
-		"spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.destination=some_other_stream,"
+		"spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.multiplex = true",
+		"spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.destination = some_other_stream,"
 				+ KinesisBinderFunctionalTests.KINESIS_STREAM,
 		"spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.idleBetweenPolls = 1",
 		"spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.listenerMode = batch",
 		"spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.checkpointMode = manual",
 		"spring.cloud.stream.kinesis.binder.headers = event.eventType",
 		"spring.cloud.stream.kinesis.binder.autoAddShards = true" })
 @DirtiesContext
-@Disabled("Something is off with generics in Spring Cloud Stream for batch processing")
 public class KinesisBinderFunctionalTests implements LocalstackContainerTest {
 
 	static final String KINESIS_STREAM = "test_stream";
@@ -171,7 +169,7 @@ public ObjectMapper objectMapper() {
 		}
 
 		@Bean
-		public AtomicReference<Message<List<Message<?>>>> messageHolder() {
+		public AtomicReference<Message<?>> messageHolder() {
 			return new AtomicReference<>();
 		}
 
@@ -181,8 +179,8 @@ public CountDownLatch messageBarrier() {
 		}
 
 		@Bean
-		public Consumer<Message<List<Message<?>>>> eventConsumerBatchProcessingWithHeaders(
-				AtomicReference<Message<List<Message<?>>>> messageHolder, CountDownLatch messageBarrier) {
+		public Consumer<Message<?>> eventConsumerBatchProcessingWithHeaders(AtomicReference<Message<?>> messageHolder,
+				CountDownLatch messageBarrier) {
 
 			return eventMessages -> {
 				messageHolder.set(eventMessages);