private void parseStoppingOffsets(
      KafkaPartitionSplit split,
      List<TopicPartition> partitionsStoppingAtLatest,
      Set<TopicPartition> partitionsStoppingAtCommitted) {
    TopicPartition tp = split.getTopicPartition();
    split
        .getStoppingOffset()
        .ifPresent(
            stoppingOffset -> {
              if (stoppingOffset >= 0) {
                stoppingOffsets.put(tp, stoppingOffset);
              } else if (stoppingOffset == KafkaPartitionSplit.LATEST_OFFSET) {
                partitionsStoppingAtLatest.add(tp);
              } else if (stoppingOffset == KafkaPartitionSplit.COMMITTED_OFFSET) {
                partitionsStoppingAtCommitted.add(tp);
              } else {
                // This should not happen.
                throw new FlinkRuntimeException(
                    String.format(
                        "Invalid stopping offset %d for partition %s", stoppingOffset, 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