public void clearEpicStartOffsetsBeforeOrEqual(
      TopicPartition tp, String upstreamId, long epicNo) {
    String key = combineTopicPartitionAndUpstreamIdAndEpicNo(tp, upstreamId, epicNo);
    NavigableMap<String, Long> beforeOrEqual = upstreamEpicStartOffsets.headMap(key, true);

    String prefix = combineTopicPartitionAndUpstreamId(tp, upstreamId);
    for (String s : beforeOrEqual.keySet()) {
      if (!s.contains(prefix)) {
        continue;
      }
      upstreamEpicStartOffsets.remove(s);
    }
  }
clear the epic start offsets before or equal the epicNo in the topicPartition.

@param tp
@param upstreamId
@param epicNo
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/LogSourceHelper.java