@Nullable
    @Override
    public ConsumerRecord<byte[], byte[]> nextRecordFromSplit() {
      Preconditions.checkNotNull(
          currentTopicPartition,
          "Make sure nextSplit() did not return null before iterate over the records split.");
      if (recordIterator.hasNext()) {
        final ConsumerRecord<byte[], byte[]> record = recordIterator.next();
        // Only emit records before stopping offset
        if (record.offset() < currentSplitStoppingOffset) {
          metrics.recordCurrentOffset(currentTopicPartition, record.offset());
          return record;
        }
      }
      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