private static void snapshotMetricStateForSourceSubMetricData(SourceMetricData sourceMetricData,
            Integer subtaskIndex, MetricState metricState) {
        if (!(sourceMetricData instanceof SourceSubMetricData)) {
            return;
        }
        SourceSubMetricData sourcesubMetricData = (SourceSubMetricData) sourceMetricData;
        // snapshot read phase metric
        Map<String, Long> metricDataMap = metricState.getMetrics();
        Map<ReadPhase, ReadPhaseMetricData> readPhaseMetricMap = sourcesubMetricData.getReadPhaseMetricMap();
        if (readPhaseMetricMap != null && !readPhaseMetricMap.isEmpty()) {
            Set<Entry<ReadPhase, ReadPhaseMetricData>> entries = readPhaseMetricMap.entrySet();
            for (Entry<ReadPhase, ReadPhaseMetricData> entry : entries) {
                Counter readPhase = entry.getValue().getReadPhase();
                metricDataMap.put(entry.getKey().getPhase(), (readPhase != null) ? readPhase.getCount() : 0L);
            }
        }
        // snapshot sub source metric
        Map<String, SourceMetricData> subSourceMetricMap = sourcesubMetricData.getSubSourceMetricMap();
        if (subSourceMetricMap != null && !subSourceMetricMap.isEmpty()) {
            Map<String, MetricState> subMetricStateMap = new HashMap<>();
            Set<Entry<String, SourceMetricData>> entries = subSourceMetricMap.entrySet();
            for (Entry<String, SourceMetricData> entry : entries) {
                Map<String, Long> subMetricDataMap = new HashMap<>(4);
                subMetricDataMap.put(NUM_RECORDS_IN, entry.getValue().getNumRecordsIn().getCount());
                subMetricDataMap.put(NUM_BYTES_IN, entry.getValue().getNumBytesIn().getCount());
                subMetricStateMap.put(entry.getKey(), new MetricState(subtaskIndex, subMetricDataMap));
            }
            metricState.setSubMetricStateMap(subMetricStateMap);
        }
    }
Snapshot sub metric state data for {@link SourceSubMetricData}
@param sourceMetricData {@link SourceMetricData} A collection class for handling metrics
@param subtaskIndex subtask index
@param metricState state of source metric data
@throws Exception throw exception when add sub metric state
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/util/MetricStateUtils.java