@Override
    public List<SourceSplitBase> snapshotState(long checkpointId) {
        // unfinished splits
        List<SourceSplitBase> stateSplits = super.snapshotState(checkpointId);

        // add finished snapshot splits that didn't receive ack yet
        stateSplits.addAll(finishedUnackedSplits.values());

        // add stream splits who are uncompleted
        stateSplits.addAll(uncompletedStreamSplits.values());

        SourceTableMetricData sourceMetricData = sourceReaderMetrics.getSourceMetricData();
        LOG.info("metric-states snapshot sourceMetricData:{}", sourceMetricData);
        if (sourceMetricData != null) {
            long countNumBytesIn = sourceMetricData.getNumBytesIn().getCount();
            long countNumRecordsIn = sourceMetricData.getNumRecordsIn().getCount();
            Map<String, Long> readPhaseMetricMap = sourceMetricData.getReadPhaseMetricMap().entrySet().stream().collect(
                    Collectors.toMap(v -> v.getKey().getPhase(), e -> e.getValue().getReadPhase().getCount()));
            Map<String, TableMetric> tableMetricMap = sourceMetricData.getSubSourceMetricMap().entrySet().stream()
                    .collect(Collectors.toMap(Entry::getKey,
                            e -> new TableMetric(e.getValue().getNumRecordsIn().getCount(),
                                    e.getValue().getNumBytesIn().getCount())));
            stateSplits
                    .add(new MetricSplit(countNumBytesIn, countNumRecordsIn, readPhaseMetricMap, tableMetricMap));
        }

        return stateSplits;
    }
The multi-parallel source reader for table snapshot phase from {@link SnapshotSplit} and then
single-parallel source reader for table stream phase from {@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/IncrementalSourceReader.java