private void addMySqlMetricSplit(List<MySqlSplit> unfinishedSplits) {
        SourceTableMetricData sourceMetricData = sourceReaderMetrics.getSourceMetricData();
        LOG.info("inlong-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, MySqlTableMetric> tableMetricMap = sourceMetricData.getSubSourceMetricMap().entrySet().stream()
                    .collect(Collectors.toMap(Entry::getKey,
                            e -> new MySqlTableMetric(e.getValue().getNumRecordsIn().getCount(),
                                    e.getValue().getNumBytesIn().getCount())));
            unfinishedSplits
                    .add(new MySqlMetricSplit(countNumBytesIn, countNumRecordsIn, readPhaseMetricMap, tableMetricMap));
        }
    }
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