public static List<Integer> computeIndexList(Integer subtaskIndex, Integer currentSubtaskNum,
            Integer previousSubtaskNum) {
        List<Integer> indexList = new ArrayList<>();
        int assignTaskNum = previousSubtaskNum / currentSubtaskNum;
        if (subtaskIndex == currentSubtaskNum - 1) {
            for (int i = subtaskIndex * assignTaskNum; i < previousSubtaskNum; i++) {
                indexList.add(i);
            }
        } else {
            for (int i = 1; i <= assignTaskNum; i++) {
                indexList.add(i + subtaskIndex * assignTaskNum - 1);
            }
        }
        return indexList;
    }
Assignment previous subtask index to current subtask when reduce parallelism
n = N/m, get n old task per new subtask, mth new subtask get (N - (m - 1) * n) old task
@param subtaskIndex current subtask index
@param currentSubtaskNum number of current parallel subtask
@param previousSubtaskNum number of previous parallel subtask
@return index list
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/util/MetricStateUtils.java