@Override
    public RecordsWithSplitIds<SourceRecord> fetch() throws IOException {

        checkSplitOrStartNext();
        checkNeedStopBinlogReader();

        Iterator<SourceRecord> dataIt;
        try {
            dataIt = currentReader.pollSplitRecords();
        } catch (InterruptedException e) {
            LOG.warn("fetch data failed.", e);
            throw new IOException(e);
        }
        return dataIt == null
                ? finishedSnapshotSplit()
                : MySqlRecords.forRecords(currentSplitId, dataIt);
    }
The {@link SplitReader} implementation for the {@link MySqlSource}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSplitReader.java