private void parseStartingOffsets(
      KafkaPartitionSplit split,
      List<TopicPartition> partitionsStartingFromEarliest,
      List<TopicPartition> partitionsStartingFromLatest,
      Map<TopicPartition, Long> partitionsStartingFromSpecifiedOffsets) {
    TopicPartition tp = split.getTopicPartition();
    // Parse starting offsets.
    if (split.getStartingOffset() == KafkaPartitionSplit.EARLIEST_OFFSET) {
      partitionsStartingFromEarliest.add(tp);
    } else if (split.getStartingOffset() == KafkaPartitionSplit.LATEST_OFFSET) {
      partitionsStartingFromLatest.add(tp);
    } else if (split.getStartingOffset() == KafkaPartitionSplit.COMMITTED_OFFSET) {
      // Do nothing here, the consumer will first try to get the committed offsets of
      // these partitions by default.
    } else {
      partitionsStartingFromSpecifiedOffsets.put(tp, split.getStartingOffset());
    }
  }
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