protected void checkSplitOrStartNext() throws IOException {
        // the stream fetcher should keep alive
        if (currentFetcher instanceof IncrementalSourceStreamFetcher) {
            return;
        }

        if (canAssignNextSplit()) {
            final SourceSplitBase nextSplit = splits.poll();
            if (nextSplit == null) {
                throw new IOException("Cannot fetch from another split - no split remaining.");
            }
            currentSplitId = nextSplit.splitId();

            if (nextSplit.isSnapshotSplit()) {
                if (currentFetcher == null) {
                    final FetchTask.Context taskContext =
                            dataSourceDialect.createFetchTaskContext(nextSplit, sourceConfig);
                    currentFetcher = new IncrementalSourceScanFetcher(taskContext, subtaskId);
                }
            } else {
                // point from snapshot split to stream split
                if (currentFetcher != null) {
                    LOG.info("It's turn to read stream split, close current snapshot fetcher.");
                    currentFetcher.close();
                }
                final FetchTask.Context taskContext =
                        dataSourceDialect.createFetchTaskContext(nextSplit, sourceConfig);
                currentFetcher = new IncrementalSourceStreamFetcher(taskContext, subtaskId);
                LOG.info("Stream fetcher is created.");
            }
            currentFetcher.submitTask(dataSourceDialect.createFetchTask(nextSplit));
        }
    }
Basic class read {@link SourceSplitBase} and return {@link SourceRecord}.
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/IncrementalSourceSplitReader.java