[FLINK-39660][Connectors/Kinesis] Fix Netty event loop threads blocking behavior and concurrent activation race in FanOutKinesisShardSubscription

Author: chenylee-aws

flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java

@@ -72,7 +72,6 @@
 import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
 import software.amazon.awssdk.services.kinesis.model.Record;
 import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException;
-import software.amazon.awssdk.utils.AttributeMap;
 
 import java.time.Duration;
 import java.util.Map;
@@ -271,7 +270,7 @@ private KinesisAsyncStreamProxy createKinesisAsyncStreamProxy(
 
         SdkAsyncHttpClient asyncHttpClient =
                 AWSGeneralUtil.createAsyncHttpClient(
-                        AttributeMap.builder().build(), NettyNioAsyncHttpClient.builder());
+                        kinesisClientProperties, NettyNioAsyncHttpClient.builder());
         KinesisAsyncClient kinesisAsyncClient =
                 AWSClientUtil.createAwsAsyncClient(
                         kinesisClientProperties,

flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/fanout/FanOutKinesisShardSplitReader.java

@@ -93,6 +93,7 @@ public void handleSplitsChanges(SplitsChange<KinesisShardSplit> splitsChanges) {
 
     @Override
     public void close() throws Exception {
+        splitSubscriptions.values().forEach(FanOutKinesisShardSubscription::close);
         asyncStreamProxy.close();
     }
 }