@Override
  public RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> fetch() throws IOException {
    KafkaPartitionSplitRecords recordsBySplits;
    Set<TopicPartition> retractTps;
    if (logRetractionEnable
        && !(retractTps = logReadHelper.getRetractTopicPartitions()).isEmpty()) {
      recordsBySplits = readReversely(retractTps);
    } else {
      recordsBySplits = readNormal();
    }

    return recordsBySplits;
  }
This reader supports read log data in log-store. If {@link
MixedFormatValidator#MIXED_FORMAT_LOG_CONSISTENCY_GUARANTEE_ENABLE} values true, reader would
read data consistently with file-store. Some data would be written into log-store repeatedly if
upstream job failovers several times, so it's necessary to retract these data to guarantee the
consistency with file-store.

<pre>
The data in log-store with Flip like: 1 2 3 4 5   6 7 8 9  Flip  6 7 8 9 10 11 12   13 14
                                      ckp-1     |ckp-2   |     | ckp-2            | ckp-3
The data reads like: 1 2 3 4 5 6 7 8 9 -9 -8 -7 -6 6 7 8 9 10 11 12 13 14

The implementation of reading consistently lists below:
1. read data normally {@link #readNormal()}
   - convert data to {@link LogRecordWithRetractInfo} in {@link #convertToLogRecord(ConsumerRecords)}. If it comes to
   Flip, the data would be cut.
   - save retracting info {@link LogSourceHelper.EpicRetractingInfo} in
   {@link LogSourceHelper#startRetracting(TopicPartition, String, long, long)}.
   - record the epic start offsets
   {@link LogSourceHelper#initialEpicStartOffsetIfEmpty(TopicPartition, String, long, long)} in
   - handle normal data like {@link KafkaPartitionSplitReader}
2. read data reversely {@link #readReversely} if some topic partitions come into Flip,
 i.e. {@link LogSourceHelper#getRetractTopicPartitions()}
   - record the offsets that consumer's current positions, stoppingOffsetsFromConsumer.
   - reset consumer to the offset: current position - batchSize
   - poll data until stoppingOffsetsFromConsumer {@link #pollToDesignatedPositions}
   - locate the stop offset in the batch data {@link #findIndexOfOffset(List, long)}, and start from it to read
   reversely, stop at {@link LogSourceHelper.EpicRetractingInfo#getRetractStoppingOffset()}
   - suspend retract {@link LogSourceHelper#suspendRetracting(TopicPartition)} when it comes to
   {@link LogSourceHelper.EpicRetractingInfo#getRetractStoppingOffset()}, else repeat {@link #readReversely} in next
   {@link #fetch()}
3. write offset and retract info into splitState in
{@link LogKafkaPartitionSplitState#updateState(LogRecordWithRetractInfo)}
4. initialize state from state {@link LogSourceHelper#initializedState}
</pre>
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/kafka/LogKafkaPartitionSplitReader.java