protected void markEmptySplitsAsFinished(KafkaPartitionSplitRecords recordsBySplits) {
    // Some splits are discovered as empty when handling split additions. These splits should be
    // added to finished splits to clean up states in split fetcher and source reader.
    if (!emptySplits.isEmpty()) {
      recordsBySplits.finishedSplits.addAll(emptySplits);
      emptySplits.clear();
    }
  }
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