[FLINK-36630][Connectors/Kafka] Wrap consumer.position in retryOnWakeup (#133)

Author: trompa

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java

@@ -124,7 +124,7 @@ public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOExce
         List<TopicPartition> finishedPartitions = new ArrayList<>();
         for (TopicPartition tp : consumer.assignment()) {
             long stoppingOffset = getStoppingOffset(tp);
-            long consumerPosition = consumer.position(tp);
+            long consumerPosition = getConsumerPosition(tp, "retrieving consumer position");
             // Stop fetching when the consumer's position reaches the stoppingOffset.
             // Control messages may follow the last record; therefore, using the last record's
             // offset as a stopping condition could result in indefinite blocking.
@@ -279,6 +279,10 @@ void setConsumerClientRack(Properties consumerProps, String rackId) {
         }
     }
 
+    long getConsumerPosition(TopicPartition tp, String msg) {
+        return retryOnWakeup(() -> consumer.position(tp), msg);
+    }
+
     private void parseStartingOffsets(
             KafkaPartitionSplit split,
             List<TopicPartition> partitionsStartingFromEarliest,
@@ -371,10 +375,9 @@ private void removeEmptySplits() {
         List<TopicPartition> emptyPartitions = new ArrayList<>();
         // If none of the partitions have any records,
         for (TopicPartition tp : consumer.assignment()) {
-            if (retryOnWakeup(
-                            () -> consumer.position(tp),
-                            "getting starting offset to check if split is empty")
-                    >= getStoppingOffset(tp)) {
+            long startingOffset =
+                    getConsumerPosition(tp, "getting starting offset to check if split is empty");
+            if (startingOffset >= getStoppingOffset(tp)) {
                 emptyPartitions.add(tp);
             }
         }
@@ -403,9 +406,8 @@ private void maybeLogSplitChangesHandlingResult(
                 }
 
                 long startingOffset =
-                        retryOnWakeup(
-                                () -> consumer.position(split.getTopicPartition()),
-                                "logging starting position");
+                        getConsumerPosition(split.getTopicPartition(), "logging starting position");
+
                 long stoppingOffset = getStoppingOffset(split.getTopicPartition());
                 splitsInfo.add(
                         String.format(