private Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> pollToDesignatedPositions(
      Map<TopicPartition, Long> stoppingOffsets) {
    ConsumerRecords<byte[], byte[]> consumerRecords;
    try {
      consumerRecords = consumer.poll(Duration.ofMillis(POLL_TIMEOUT));
    } catch (WakeupException we) {
      LOG.error("consume reversely error");
      return Collections.EMPTY_MAP;
    }

    Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordsForTps = new HashMap<>();

    int unfinished = stoppingOffsets.size();
    int round = 0;

    Set<TopicPartition> unfinishedTps = new HashSet<>();
    while (unfinished > 0 && round++ < RETRACT_FETCH_MAX_ROUND) {
      unfinishedTps.clear();

      for (TopicPartition tp : consumerRecords.partitions()) {
        recordsForTps.putIfAbsent(tp, new ArrayList<>(RETRACT_SIZE));
        List<ConsumerRecord<byte[], byte[]>> records = recordsForTps.get(tp);

        records.addAll(consumerRecords.records(tp));

        long stoppingOffset = stoppingOffsets.get(tp);
        if (records.get(records.size() - 1).offset() >= stoppingOffset) {
          unfinished--;
          LOG.info(
              "reach the stopping offset. stopping offset: {}, tp: {}. data size:{}",
              stoppingOffset,
              tp,
              records.size());
        } else {
          unfinishedTps.add(tp);
        }
      }
      if (unfinished == 0) {
        break;
      }
      consumer.assign(unfinishedTps);
    }

    if (unfinished > 0) {
      LOG.error("can not poll msg to designated positions. unfinished: {}", unfinishedTps);
      for (TopicPartition tp : unfinishedTps) {
        List<ConsumerRecord<byte[], byte[]>> records = recordsForTps.get(tp);
        LOG.info(
            "tp: {}, polled offset:{}, stopping offset: {}",
            tp,
            records.get(records.size() - 1).offset(),
            stoppingOffsets.get(tp));
      }
      throw new UnsupportedOperationException("poll msg reversely error");
    }

    return recordsForTps;
  }
@param stoppingOffsets the stopping offset is the position which should be read.
@return value in map may contain some useless records. It should be filtered.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/kafka/LogKafkaPartitionSplitReader.java