public void initializedState(KafkaPartitionSplit s) {
    if (!(s instanceof LogKafkaPartitionSplit)) {
      return;
    }
    LogKafkaPartitionSplit split = (LogKafkaPartitionSplit) s;
    if (split.isRetracting()) {
      retractingInfo.put(
          split.getTopicPartition(),
          EpicRetractingInfo.of(
              split.getRetractingEpicNo(), split.getRetractingUpstreamId(),
              split.getRetractStopOffset(), split.getRevertStartOffset()));
    }
    Map<String, Long> upStreamEpicStartOffsets = split.getUpStreamEpicStartOffsets();

    upStreamEpicStartOffsets.forEach(
        (upstreamEpic, offset) -> {
          String key =
              combineTopicPartitionAndUpstreamIdAndEpicNo(split.getTopicPartition(), upstreamEpic);
          upstreamEpicStartOffsets.putIfAbsent(key, offset);
        });
  }
Key: topic partition + "_" + upstream job id + "_" + epicNo, generated by {@link
#combineTopicPartitionAndUpstreamIdAndEpicNo)} method. Value: epic start offset
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/LogSourceHelper.java