private void seekToStartingOffsets(
      List<TopicPartition> partitionsStartingFromEarliest,
      List<TopicPartition> partitionsStartingFromLatest,
      Map<TopicPartition, Long> partitionsStartingFromSpecifiedOffsets) {

    if (!partitionsStartingFromEarliest.isEmpty()) {
      LOG.trace("Seeking starting offsets to beginning: {}", partitionsStartingFromEarliest);
      consumer.seekToBeginning(partitionsStartingFromEarliest);
    }

    if (!partitionsStartingFromLatest.isEmpty()) {
      LOG.trace("Seeking starting offsets to end: {}", partitionsStartingFromLatest);
      consumer.seekToEnd(partitionsStartingFromLatest);
    }

    if (!partitionsStartingFromSpecifiedOffsets.isEmpty()) {
      LOG.trace(
          "Seeking starting offsets to specified offsets: {}",
          partitionsStartingFromSpecifiedOffsets);
      partitionsStartingFromSpecifiedOffsets.forEach(consumer::seek);
    }
  }
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