@Nullable
    @Override
    public Iterator<SourceRecords> pollSplitRecords() throws InterruptedException {
        checkReadException();

        if (hasNextElement.get()) {
            // eg:
            // data input: [low watermark event][snapshot events][high watermark event][change
            // events][end watermark event]
            // data output: [low watermark event][normalized events][high watermark event]
            boolean reachChangeLogStart = false;
            boolean reachChangeLogEnd = false;
            SourceRecord lowWatermark = null;
            SourceRecord highWatermark = null;
            Map<Struct, SourceRecord> outputBuffer = new HashMap<>();
            List<SourceRecord> snapshotRecordsWithoutKey = new ArrayList<>();
            while (!reachChangeLogEnd) {
                checkReadException();
                List<DataChangeEvent> batch = queue.poll();
                for (DataChangeEvent event : batch) {
                    SourceRecord record = event.getRecord();
                    if (lowWatermark == null) {
                        lowWatermark = record;
                        assertLowWatermark(lowWatermark);
                        continue;
                    }

                    if (highWatermark == null && isHighWatermarkEvent(record)) {
                        highWatermark = record;
                        // snapshot events capture end and begin to capture binlog events
                        reachChangeLogStart = true;
                        continue;
                    }

                    if (reachChangeLogStart && isEndWatermarkEvent(record)) {
                        // capture to end watermark events, stop the loop
                        reachChangeLogEnd = true;
                        break;
                    }

                    if (!reachChangeLogStart) {
                        if (record.key() == null) {
                            snapshotRecordsWithoutKey.add(record);
                        } else {
                            outputBuffer.put((Struct) record.key(), record);
                        }
                    } else {
                        if (isChangeRecordInChunkRange(record)) {
                            // rewrite overlapping snapshot records through the record key
                            taskContext.rewriteOutputBuffer(outputBuffer, record);
                        }
                    }
                }
            }
            // snapshot split return its data once
            hasNextElement.set(false);

            // split
            final List<SourceRecord> normalizedRecords = new ArrayList<>();
            normalizedRecords.add(lowWatermark);
            if (snapshotRecordsWithoutKey.isEmpty()) {
                normalizedRecords.addAll(taskContext.formatMessageTimestamp(outputBuffer.values()));
            } else {
                normalizedRecords.addAll(taskContext.formatMessageTimestamp(snapshotRecordsWithoutKey));
            }
            normalizedRecords.add(highWatermark);

            final List<SourceRecords> sourceRecordsSet = new ArrayList<>();
            sourceRecordsSet.add(new SourceRecords(normalizedRecords));
            return sourceRecordsSet.iterator();
        }
        // the data has been polled, no more data
        reachEnd.compareAndSet(false, true);
        return null;
    }
Fetcher to fetch data from table split, the split is the snapshot split {@link SnapshotSplit}.
Copy from com.ververica:flink-cdc-base:2.3.0.
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java