public void updateState(LogRecordWithRetractInfo<RowData> record) {
    if (record.isRetracting()) {
      setCurrentOffset(record.offset() - 1);
      revertStartOffset = record.getRevertStartingOffset();
      retractStopOffset = record.getRetractStoppingOffset();
      retractingEpicNo = record.getRetractingEpicNo();
      retractingUpstreamId = record.getLogData().getUpstreamId();
    } else {
      setCurrentOffset(record.offset() + 1);
    }
    initEpicStartOffsetIfEmpty(
        record.getLogData().getUpstreamId(), record.getLogData().getEpicNo(), record.offset());

    // todo: clear useless epic start offset in state
    retracting = record.isRetracting();
  }
Key: upstream job id + "_" + epicNo, Value: epic start offset
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/kafka/LogKafkaPartitionSplitState.java