@Override
    public void addSplits(List<MySqlSplit> splits) {
        // restore for finishedUnackedSplits
        List<MySqlSplit> unfinishedSplits = new ArrayList<>();
        for (MySqlSplit split : splits) {
            LOG.info("Add Split: " + split);
            if (split.isMetricSplit()) {
                MySqlMetricSplit mysqlMetricSplit = (MySqlMetricSplit) split;
                LOG.info("inlong-metric-states restore metricSplit:{}", mysqlMetricSplit);
                sourceReaderMetrics.initMetrics(mysqlMetricSplit.getNumRecordsIn(),
                        mysqlMetricSplit.getNumBytesIn(), mysqlMetricSplit.getReadPhaseMetricMap(),
                        mysqlMetricSplit.getTableMetricMap());
                LOG.info("inlong-metric-states restore sourceReaderMetrics:{}",
                        sourceReaderMetrics.getSourceMetricData());
                continue;
            }
            if (split.isSnapshotSplit()) {
                MySqlSnapshotSplit snapshotSplit = split.asSnapshotSplit();
                if (snapshotSplit.isSnapshotReadFinished()) {
                    finishedUnackedSplits.put(snapshotSplit.splitId(), snapshotSplit);
                } else {
                    unfinishedSplits.add(split);
                }
            } else {
                MySqlBinlogSplit binlogSplit = split.asBinlogSplit();
                // the binlog split is suspended
                if (binlogSplit.isSuspended()) {
                    suspendedBinlogSplit = binlogSplit;
                } else if (!binlogSplit.isCompletedSplit()) {
                    uncompletedBinlogSplits.put(split.splitId(), split.asBinlogSplit());
                    requestBinlogSplitMetaIfNeeded(split.asBinlogSplit());
                } else {
                    uncompletedBinlogSplits.remove(split.splitId());
                    MySqlBinlogSplit mySqlBinlogSplit =
                            discoverTableSchemasForBinlogSplit(split.asBinlogSplit());
                    mySqlBinlogSplit.getFinishedSnapshotSplitInfos()
                            .sort(Comparator.comparingInt(splitInfo -> chunkId(splitInfo.getSplitId())));
                    unfinishedSplits.add(mySqlBinlogSplit);
                }
            }
        }
        // notify split enumerator again about the finished unacked snapshot splits
        reportFinishedSnapshotSplitsIfNeed();
        // add all un-finished splits (including binlog split) to SourceReaderBase
        if (!unfinishedSplits.isEmpty()) {
            super.addSplits(unfinishedSplits);
        }
    }
The source reader for MySQL source splits.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlSourceReader.java