public void startRetracting(
      TopicPartition tp, String upstreamId, long epicNo, long revertStartingOffset) {
    String key = combineTopicPartitionAndUpstreamIdAndEpicNo(tp, upstreamId, epicNo);
    if (!upstreamEpicStartOffsets.containsKey(key)) {
      // data have not been read, so that it's unnecessary to retract
      return;
    }
    long retractStoppingOffset = upstreamEpicStartOffsets.get(key);

    retractingInfo.put(
        tp,
        new EpicRetractingInfo(epicNo, upstreamId, retractStoppingOffset, revertStartingOffset));
  }
@param revertStartingOffset the offset where job revert to normal read starts from. It should
    skip the flip which has been read.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/LogSourceHelper.java