public void revertConsumer(Set<TopicPartition> finishRetract) {
    for (TopicPartition tp : finishRetract) {
      LogSourceHelper.EpicRetractingInfo retractingInfo = logReadHelper.getRetractInfo(tp);
      long revert = retractingInfo.getRevertStartingOffset();
      consumer.seek(tp, revert);
    }
  }
revert consumer to original offset after flip
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/kafka/LogKafkaPartitionSplitReader.java