@Override
    public void addSplits(List<SourceSplitBase> splits) {
        // restore for finishedUnackedSplits
        List<SourceSplitBase> unfinishedSplits = new ArrayList<>();
        for (SourceSplitBase split : splits) {
            if (split instanceof MetricSplit) {
                MetricSplit metricSplit = (MetricSplit) split;
                LOG.info("inlong-metric-states restore metricSplit:{}", metricSplit);
                sourceReaderMetrics.initMetrics(metricSplit.getNumRecordsIn(),
                        metricSplit.getNumBytesIn(), metricSplit.getReadPhaseMetricMap(),
                        metricSplit.getTableMetricMap());
                LOG.info("inlong-metric-states restore sourceReaderMetrics:{}",
                        sourceReaderMetrics.getSourceMetricData());
                continue;
            }
            if (split.isSnapshotSplit()) {
                SnapshotSplit snapshotSplit = split.asSnapshotSplit();
                if (snapshotSplit.isSnapshotReadFinished()) {
                    finishedUnackedSplits.put(snapshotSplit.splitId(), snapshotSplit);
                } else {
                    unfinishedSplits.add(split);
                }
            } else {
                // the stream split is uncompleted
                if (!split.asStreamSplit().isCompletedSplit()) {
                    uncompletedStreamSplits.put(split.splitId(), split.asStreamSplit());
                    requestStreamSplitMetaIfNeeded(split.asStreamSplit());
                } else {
                    uncompletedStreamSplits.remove(split.splitId());
                    StreamSplit streamSplit =
                            discoverTableSchemasForStreamSplit(split.asStreamSplit());
                    unfinishedSplits.add(streamSplit);
                }
            }
        }
        // notify split enumerator again about the finished unacked snapshot splits
        reportFinishedSnapshotSplitsIfNeed();
        // add all un-finished splits (including stream split) to SourceReaderBase
        super.addSplits(unfinishedSplits);
    }
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