@Override
    public IncrementalSourceReader<T, C> createReader(SourceReaderContext readerContext)
            throws Exception {
        // create source config for the given subtask (e.g. unique server id)
        C sourceConfig = configFactory.create(readerContext.getIndexOfSubtask());
        FutureCompletingBlockingQueue<RecordsWithSplitIds<SourceRecords>> elementsQueue =
                new FutureCompletingBlockingQueue<>();

        // Forward compatible with flink 1.13
        final Method metricGroupMethod = readerContext.getClass().getMethod("metricGroup");
        metricGroupMethod.setAccessible(true);
        final MetricGroup metricGroup = (MetricGroup) metricGroupMethod.invoke(readerContext);
        final SourceReaderMetrics sourceReaderMetrics = new SourceReaderMetrics(metricGroup);

        sourceReaderMetrics.registerMetrics();
        Supplier<IncrementalSourceSplitReader<C>> splitReaderSupplier =
                () -> new IncrementalSourceSplitReader<>(
                        readerContext.getIndexOfSubtask(), dataSourceDialect, sourceConfig);
        return new IncrementalSourceReader<>(
                elementsQueue,
                splitReaderSupplier,
                createRecordEmitter(sourceConfig, sourceReaderMetrics),
                readerContext.getConfiguration(),
                readerContext,
                sourceConfig,
                sourceSplitSerializer,
                dataSourceDialect);
    }
The basic source of Incremental Snapshot framework for datasource, it is based on FLIP-27 and
Watermark Signal Algorithm which supports parallel reading snapshot of table and then continue to
capture data change by streaming reading.
<p>
Copy from com.ververica:flink-connector-mongodb-cdc-2.3.0
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/mongodb/source/IncrementalSource.java