public static void snapshotMetricStateForSourceMetricData(ListState<MetricState> metricStateListState,
            SourceMetricData sourceMetricData, Integer subtaskIndex)
            throws Exception {
        LOGGER.info("snapshotMetricStateForSourceMetricData:{}, sourceMetricData:{}, subtaskIndex:{}",
                metricStateListState, sourceMetricData, subtaskIndex);
        metricStateListState.clear();
        Map<String, Long> metricDataMap = new HashMap<>();
        metricDataMap.put(NUM_RECORDS_IN, sourceMetricData.getNumRecordsIn().getCount());
        metricDataMap.put(NUM_BYTES_IN, sourceMetricData.getNumBytesIn().getCount());
        MetricState metricState = new MetricState(subtaskIndex, metricDataMap);

        // snapshot sub metric data state
        snapshotMetricStateForSourceSubMetricData(sourceMetricData, subtaskIndex, metricState);
        metricStateListState.add(metricState);
    }
Snapshot metric state data for {@link SourceMetricData}
@param metricStateListState state data list
@param sourceMetricData {@link SourceMetricData} A collection class for handling metrics
@param subtaskIndex subtask index
@throws Exception throw exception when add metric state
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/util/MetricStateUtils.java