private KafkaPartitionSplitRecords readNormal() throws IOException {
    ConsumerRecords<byte[], byte[]> consumerRecords;
    try {
      consumerRecords = consumer.poll(Duration.ofMillis(POLL_TIMEOUT));
    } catch (WakeupException | IllegalStateException e) {
      // IllegalStateException will be thrown if the consumer is not assigned any partitions.
      // This happens if all assigned partitions are invalid or empty (starting offset >=
      // stopping offset). We just mark empty partitions as finished and return an empty
      // record container, and this consumer will be closed by SplitFetcherManager.
      KafkaPartitionSplitRecords recordsBySplits =
          new KafkaPartitionSplitRecords(ConsumerRecords.empty(), kafkaSourceReaderMetrics);
      markEmptySplitsAsFinished(recordsBySplits);
      return recordsBySplits;
    }

    ConsumerRecords<byte[], byte[]> logRecords = convertToLogRecord(consumerRecords);
    KafkaPartitionSplitRecords recordsBySplits =
        new KafkaPartitionSplitRecords(logRecords, kafkaSourceReaderMetrics);

    List<TopicPartition> finishedPartitions = new ArrayList<>();
    for (TopicPartition tp : logRecords.partitions()) {
      long stoppingOffset = getStoppingOffset(tp);
      final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition = logRecords.records(tp);

      if (recordsFromPartition.size() > 0) {
        final ConsumerRecord<byte[], byte[]> lastRecord =
            recordsFromPartition.get(recordsFromPartition.size() - 1);

        // After processing a record with offset of "stoppingOffset - 1", the split reader
        // should not continue fetching because the record with stoppingOffset may not
        // exist. Keep polling will just block forever.
        if (lastRecord.offset() >= stoppingOffset - 1) {
          recordsBySplits.setPartitionStoppingOffset(tp, stoppingOffset);
          finishSplitAtRecord(
              tp, stoppingOffset, lastRecord.offset(), finishedPartitions, recordsBySplits);
        }
      }
      // Track this partition's record lag if it never appears before
      kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, tp);
    }

    markEmptySplitsAsFinished(recordsBySplits);

    // Unassign the partitions that has finished.
    if (!finishedPartitions.isEmpty()) {
      finishedPartitions.forEach(kafkaSourceReaderMetrics::removeRecordsLagMetric);
      unassignPartitions(finishedPartitions);
    }

    // Update numBytesIn
    kafkaSourceReaderMetrics.updateNumBytesInCounter();

    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