Use Flux for KCL multi stream load

* Upgrade to `com.gradle.enterprise:3.15`
* Use `setFanOut(false)` in the `KclMessageDrivenChannelAdapterTests` to cover the polling mode for KCL

Author: artembilan

settings.gradle

@@ -1,5 +1,5 @@
 plugins {
-	id 'com.gradle.enterprise' version '3.14.1'
+	id 'com.gradle.enterprise' version '3.15'
 	id 'io.spring.ge.conventions' version '0.0.13'
 }
 

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

@@ -26,12 +26,14 @@
 import javax.annotation.Nullable;
 
 import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializer;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import software.amazon.awssdk.arns.Arn;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
 import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
-import software.amazon.awssdk.services.kinesis.model.StreamDescriptionSummary;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamSummaryResponse;
 import software.amazon.awssdk.utils.BinaryUtils;
 import software.amazon.kinesis.common.ConfigsBuilder;
 import software.amazon.kinesis.common.InitialPositionInStream;
@@ -274,6 +276,7 @@ protected void onInit() {
 						this.cloudWatchClient,
 						this.workerId,
 						this.recordProcessorFactory);
+
 		this.config.lifecycleConfig().taskBackoffTimeMillis(this.consumerBackoff);
 
 		RetrievalSpecificConfig retrievalSpecificConfig;
@@ -317,7 +320,6 @@ protected void doStart() {
 					+ "because it does not make sense in case of [ListenerMode.batch].");
 		}
 
-
 		this.scheduler =
 				new Scheduler(
 						this.config.checkpointConfig(),
@@ -339,7 +341,6 @@ protected void doStart() {
 	protected void doStop() {
 		super.doStop();
 		this.scheduler.shutdown();
-
 	}
 
 	@Override
@@ -396,32 +397,28 @@ public Duration waitPeriodToDeleteFormerStreams() {
 
 				};
 
-		private List<StreamConfig> streamConfigs = null;
+		private final Flux<StreamConfig> streamConfigs =
+				Flux.fromArray(KclMessageDrivenChannelAdapter.this.streams)
+						.flatMap((streamName) ->
+								Mono.fromFuture(KclMessageDrivenChannelAdapter.this.kinesisClient
+										.describeStreamSummary(request -> request.streamName(streamName))))
+						.map(DescribeStreamSummaryResponse::streamDescriptionSummary)
+						.map((summary) ->
+								StreamIdentifier.multiStreamInstance(
+										Arn.fromString(summary.streamARN()),
+										summary.streamCreationTimestamp()
+												.getEpochSecond()))
+						.map((streamIdentifier) ->
+								new StreamConfig(streamIdentifier,
+										KclMessageDrivenChannelAdapter.this.streamInitialSequence))
+						.cache();
 
 		StreamsTracker() {
 		}
 
-		private StreamConfig buildStreamConfig(String streamName) {
-			StreamDescriptionSummary descriptionSummary =
-					KclMessageDrivenChannelAdapter.this.kinesisClient
-							.describeStreamSummary(request -> request.streamName(streamName))
-							.join().streamDescriptionSummary();
-			return new StreamConfig(StreamIdentifier.multiStreamInstance(
-					Arn.fromString(descriptionSummary.streamARN()),
-					descriptionSummary.streamCreationTimestamp().getEpochSecond()),
-					KclMessageDrivenChannelAdapter.this.streamInitialSequence);
-		}
-
 		@Override
 		public List<StreamConfig> streamConfigList() {
-			if (this.streamConfigs == null) {
-				// Lazy loading the Stream Configs, only during inquiry.
-				this.streamConfigs = new ArrayList<>();
-				Arrays.stream(KclMessageDrivenChannelAdapter.this.streams)
-						.forEach(streamName -> this.streamConfigs.add(buildStreamConfig(streamName)));
-			}
-
-			return this.streamConfigs;
+			return this.streamConfigs.collectList().block();
 		}
 
 		@Override

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

@@ -128,6 +128,7 @@ public KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter() {
 					InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON));
 			adapter.setConverter(String::new);
 			adapter.setConsumerGroup("single_stream_group");
+			adapter.setFanOut(false);
 			adapter.setBindSourceRecord(true);
 			return adapter;
 		}