private void removeEmptySplits() {
    List<TopicPartition> emptyPartitions = new ArrayList<>();
    // If none of the partitions have any records,
    for (TopicPartition tp : consumer.assignment()) {
      if (retryOnWakeup(
              () -> consumer.position(tp), "getting starting offset to check if split is empty")
          >= getStoppingOffset(tp)) {
        emptyPartitions.add(tp);
      }
    }
    if (!emptyPartitions.isEmpty()) {
      LOG.debug(
          "These assigning splits are empty and will be marked as finished in later fetch: {}",
          emptyPartitions);
      // Add empty partitions to empty split set for later cleanup in fetch()
      emptySplits.addAll(
          emptyPartitions.stream().map(KafkaPartitionSplit::toSplitId).collect(Collectors.toSet()));
      // Un-assign partitions from Kafka consumer
      unassignPartitions(emptyPartitions);
    }
  }
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