@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());
        MetricConfig metricConfig = (MetricConfig) sourceConfig;
        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);

        // create source config for the given subtask (e.g. unique server id)
        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(metricConfig.getInlongMetric())
                .withAuditAddress(metricConfig.getInlongAudit())
                .withRegisterMetric(RegisteredMetric.ALL)
                .build();

        sourceReaderMetrics.registerMetrics(metricOption, metricConfig.getMetricLabelList());
        Supplier<IncrementalSourceSplitReader<C>> splitReaderSupplier =
                () -> new IncrementalSourceSplitReader<>(
                        readerContext.getIndexOfSubtask(), dataSourceDialect, sourceConfig);
        return new IncrementalSourceReader<>(
                elementsQueue,
                splitReaderSupplier,
                createRecordEmitter(sourceConfig, sourceReaderMetrics),
                readerContext.getConfiguration(),
                readerContext,
                sourceConfig,
                sourceSplitSerializer,
                dataSourceDialect,
                sourceReaderMetrics);
    }
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.
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/IncrementalSource.java