@Override
  public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOException {
    ConsumerRecords<byte[], byte[]> consumerRecords;
    try {
      consumerRecords = consumer.poll(Duration.ofMillis(POLL_TIMEOUT));
    } catch (WakeupException | IllegalStateException e) {
      // IllegalStateException will be thrown if the consumer is not assigned any partitions.
      // This happens if all assigned partitions are invalid or empty (starting offset >=
      // stopping offset). We just mark empty partitions as finished and return an empty
      // record container, and this consumer will be closed by SplitFetcherManager.
      KafkaPartitionSplitRecords recordsBySplits =
          new KafkaPartitionSplitRecords(ConsumerRecords.empty(), kafkaSourceReaderMetrics);
      markEmptySplitsAsFinished(recordsBySplits);
      return recordsBySplits;
    }
    KafkaPartitionSplitRecords recordsBySplits =
        new KafkaPartitionSplitRecords(consumerRecords, kafkaSourceReaderMetrics);
    List<TopicPartition> finishedPartitions = new ArrayList<>();
    for (TopicPartition tp : consumerRecords.partitions()) {
      long stoppingOffset = getStoppingOffset(tp);
      final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition = consumerRecords.records(tp);

      if (recordsFromPartition.size() > 0) {
        final ConsumerRecord<byte[], byte[]> lastRecord =
            recordsFromPartition.get(recordsFromPartition.size() - 1);

        // After processing a record with offset of "stoppingOffset - 1", the split reader
        // should not continue fetching because the record with stoppingOffset may not
        // exist. Keep polling will just block forever.
        if (lastRecord.offset() >= stoppingOffset - 1) {
          recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
          finishSplitAtRecord(
              tp, stoppingOffset, lastRecord.offset(), finishedPartitions, recordsBySplits);
        }
      }
      // Track this partition's record lag if it never appears before
      kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, tp);
    }

    markEmptySplitsAsFinished(recordsBySplits);

    // Unassign the partitions that has finished.
    if (!finishedPartitions.isEmpty()) {
      finishedPartitions.forEach(kafkaSourceReaderMetrics::removeRecordsLagMetric);
      unassignPartitions(finishedPartitions);
    }

    // Update numBytesIn
    kafkaSourceReaderMetrics.updateNumBytesInCounter();

    return recordsBySplits;
  }
A {@link SplitReader} implementation that reads records from Kafka partitions.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/internals/KafkaPartitionSplitReader.java