From 4293702c9f0c912f68b3ad4c748596b9fda7bd3b Mon Sep 17 00:00:00 2001
From: Lukasz Krawiec <lkrawiec@expedia.com>
Date: Mon, 3 Mar 2025 15:38:53 -0600
Subject: [PATCH] [hotfix][Connectors/Kinesis] Pass source configuration to
 SplitFetcherManager

---
 .../flink/connector/kinesis/source/KinesisStreamsSource.java   | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java
index 5fba71c6..0a26baf7 100644
--- a/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java
+++ b/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/KinesisStreamsSource.java
@@ -167,7 +167,8 @@ public SourceReader<T, KinesisShardSplit> createReader(SourceReaderContext reade
                 new KinesisStreamsRecordEmitter<>(deserializationSchema);
         return new KinesisStreamsSourceReader<>(
                 new SingleThreadFetcherManager<>(
-                        getKinesisShardSplitReaderSupplier(sourceConfig, shardMetricGroupMap)),
+                        getKinesisShardSplitReaderSupplier(sourceConfig, shardMetricGroupMap),
+                        sourceConfig),
                 recordEmitter,
                 sourceConfig,
                 readerContext,