private int findIndexOfOffset(List<ConsumerRecord<byte[], byte[]>> records, long offset) {
    int last = records.size() - 1;
    int idx = Math.min(RETRACT_SIZE, last);

    long diff = -1;
    while (idx >= 0 && idx <= last && (diff = records.get(idx).offset() - offset) != 0) {
      if (diff > 0) {
        idx--;
      } else {
        idx++;
      }
    }
    if (diff == 0) {
      LOG.debug("start index is: {}", idx);
      return idx;
    }
    LOG.info(
        "topic: {}, partition: {}, records' offset range: [{}, {}], need to find: {}",
        records.get(0).topic(),
        records.get(0).partition(),
        records.get(0).offset(),
        records.get(last).offset(),
        offset);
    throw new IllegalStateException("can not find offset in records");
  }
@param records should be in order of kafka.
@param offset Kafka offset
@return the index in records
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/kafka/LogKafkaPartitionSplitReader.java