public static void snapshotMetricStateForSinkMetricData(ListState<MetricState> metricStateListState,
            SinkMetricData sinkMetricData, Integer subtaskIndex)
            throws Exception {
        LOGGER.info("snapshotMetricStateForSinkMetricData:{}, sinkMetricData:{}, subtaskIndex:{}",
                metricStateListState, sinkMetricData, subtaskIndex);
        metricStateListState.clear();
        Map<String, Long> metricDataMap = new HashMap<>();
        metricDataMap.put(NUM_RECORDS_OUT, sinkMetricData.getNumRecordsOut().getCount());
        metricDataMap.put(NUM_BYTES_OUT, sinkMetricData.getNumBytesOut().getCount());
        if (sinkMetricData.getDirtyRecordsOut() != null) {
            metricDataMap.put(DIRTY_RECORDS_OUT, sinkMetricData.getDirtyRecordsOut().getCount());
        }
        if (sinkMetricData.getDirtyBytesOut() != null) {
            metricDataMap.put(DIRTY_BYTES_OUT, sinkMetricData.getDirtyBytesOut().getCount());
        }
        MetricState metricState = new MetricState(subtaskIndex, metricDataMap);

        // snapshot sub metric data state
        snapshotMetricStateForSinkSubMetricData(sinkMetricData, subtaskIndex, metricState);
        metricStateListState.add(metricState);
    }
Snapshot metric state data for {@link SinkMetricData}
@param metricStateListState state data list
@param sinkMetricData {@link SinkMetricData} 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