protected void finishSplitAtRecord(
      TopicPartition tp,
      long stoppingOffset,
      long currentOffset,
      List<TopicPartition> finishedPartitions,
      KafkaPartitionSplitRecords recordsBySplits) {
    LOG.debug(
        "{} has reached stopping offset {}, current offset is {}",
        tp,
        stoppingOffset,
        currentOffset);
    finishedPartitions.add(tp);
    recordsBySplits.addFinishedSplit(KafkaPartitionSplit.toSplitId(tp));
  }
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