@Nullable
    @Override
    public String nextSplit() {
      if (splitIterator.hasNext()) {
        currentTopicPartition = splitIterator.next();
        recordIterator = consumerRecords.records(currentTopicPartition).iterator();
        currentSplitStoppingOffset =
            stoppingOffsets.getOrDefault(currentTopicPartition, Long.MAX_VALUE);
        return currentTopicPartition.toString();
      } else {
        currentTopicPartition = null;
        recordIterator = null;
        currentSplitStoppingOffset = null;
        return null;
      }
    }
Catch {@link WakeupException} in Kafka consumer call and retry the invocation on exception.

<p>This helper function handles a race condition as below:

<ol>
  <li>Fetcher thread finishes a {@link KafkaConsumer#poll(Duration)} call
  <li>Task thread assigns new splits so invokes {@link #wakeUp()}, then the wakeup is recorded
      and held by the consumer
  <li>Later fetcher thread invokes {@link #handleSplitsChanges(SplitsChange)}, and interactions
      with consumer will throw {@link WakeupException} because of the previously held wakeup in
      the consumer
</ol>

<p>Under this case we need to catch the {@link WakeupException} and retry the operation.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/internals/KafkaPartitionSplitReader.java