private void acquireAndSetStoppingOffsets(
      List<TopicPartition> partitionsStoppingAtLatest,
      Set<TopicPartition> partitionsStoppingAtCommitted) {
    Map<TopicPartition, Long> endOffset = consumer.endOffsets(partitionsStoppingAtLatest);
    stoppingOffsets.putAll(endOffset);
    if (!partitionsStoppingAtCommitted.isEmpty()) {
      retryOnWakeup(
              () -> consumer.committed(partitionsStoppingAtCommitted),
              "getting committed offset as stopping offsets")
          .forEach(
              (tp, offsetAndMetadata) -> {
                Preconditions.checkNotNull(
                    offsetAndMetadata,
                    String.format(
                        "Partition %s should stop at committed offset. "
                            + "But there is no committed offset of this partition for group %s",
                        tp, groupId));
                stoppingOffsets.put(tp, offsetAndMetadata.offset());
              });
    }
  }
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