diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
index <HASH>..<HASH> 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
@@ -96,6 +96,7 @@ public class KafkaPartitionSplitReader<T> implements SplitReader<Tuple3<T, Long,
 		try {
 			consumerRecords = consumer.poll(Duration.ofMillis(POLL_TIMEOUT));
 		} catch (WakeupException we) {
+			recordsBySplits.prepareForRead();
 			return recordsBySplits;
 		}
[FLINK-<I>][connector kafka] Ensure KafkaPartitionSplitRecords returned on consumer wakeup is properly initializes