@Override
    public void submitTask(FetchTask<SourceSplitBase> fetchTask) {
        this.streamFetchTask = fetchTask;
        this.currentStreamSplit = fetchTask.getSplit().asStreamSplit();
        configureFilter();
        taskContext.configure(currentStreamSplit);
        this.queue = taskContext.getQueue();
        executorService.submit(
                () -> {
                    try {
                        streamFetchTask.execute(taskContext);
                    } catch (Exception e) {
                        LOG.error(
                                String.format(
                                        "Execute stream read task for stream split %s fail",
                                        currentStreamSplit),
                                e);
                        readException = e;
                    }
                });
    }
Fetcher to fetch data from table split, the split is the stream split {@link StreamSplit}.
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/IncrementalSourceStreamFetcher.java