public static MetricState restoreMetricState(ListState<MetricState> metricStateListState, Integer subtaskIndex,
            Integer currentSubtaskNum) throws Exception {
        if (metricStateListState == null || metricStateListState.get() == null) {
            return null;
        }
        LOGGER.info("restoreMetricState:{}, subtaskIndex:{}, currentSubtaskNum:{}", metricStateListState, subtaskIndex,
                currentSubtaskNum);
        MetricState currentMetricState;
        Map<Integer, MetricState> map = new HashMap<>(16);
        for (MetricState metricState : metricStateListState.get()) {
            map.put(metricState.getSubtaskIndex(), metricState);
        }
        int previousSubtaskNum = map.size();
        if (currentSubtaskNum >= previousSubtaskNum) {
            currentMetricState = map.get(subtaskIndex);
        } else {
            Map<String, MetricState> subMetricStateMap = new HashMap<>();
            Map<String, Long> metrics = new HashMap<>(16);
            currentMetricState = new MetricState(subtaskIndex, metrics);
            currentMetricState.setSubMetricStateMap(subMetricStateMap);
            List<Integer> indexList = computeIndexList(subtaskIndex, currentSubtaskNum, previousSubtaskNum);
            for (Integer index : indexList) {
                MetricState metricState = map.get(index);
                for (Map.Entry<String, Long> entry : metricState.getMetrics().entrySet()) {
                    if (metrics.containsKey(entry.getKey())) {
                        metrics.put(entry.getKey(), metrics.get(entry.getKey()) + entry.getValue());
                    } else {
                        metrics.put(entry.getKey(), entry.getValue());
                    }
                }
                // restore sub source metric data state
                Map<String, MetricState> subIndexMetricStateMap = metricState.getSubMetricStateMap();
                if (subIndexMetricStateMap != null && !subIndexMetricStateMap.isEmpty()) {
                    for (Entry<String, MetricState> entry : subIndexMetricStateMap.entrySet()) {
                        MetricState subMetricState;
                        if (subMetricStateMap.containsKey(entry.getKey())) {
                            subMetricState = subMetricStateMap.get(entry.getKey());
                            Map<String, Long> subMetrics = subMetricState.getMetrics();
                            Map<String, Long> currentSubMetrics = entry.getValue().getMetrics();
                            for (Entry<String, Long> currentSubEntry : currentSubMetrics.entrySet()) {
                                if (subMetrics.containsKey(currentSubEntry.getKey())) {
                                    subMetrics.put(currentSubEntry.getKey(),
                                            subMetrics.get(currentSubEntry.getKey()) + currentSubEntry.getValue());
                                } else {
                                    subMetrics.put(currentSubEntry.getKey(), currentSubEntry.getValue());
                                }
                            }
                        } else {
                            subMetricState = entry.getValue();
                            subMetricState.setSubtaskIndex(subtaskIndex);
                        }
                        subMetricStateMap.put(entry.getKey(), subMetricState);
                    }
                }
            }
        }
        return currentMetricState;
    }
restore metric state data
@param metricStateListState state data list
@param subtaskIndex current subtask index
@param currentSubtaskNum number of current parallel subtask
@return metric state
@throws Exception throw exception metricStateListState.get()
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/util/MetricStateUtils.java