private List<FlowCtrlItem> parseLowFetchLimit(int recordNo, int typeVal,
            JsonObject jsonObject) throws Exception {
        if (jsonObject == null || jsonObject.get("type").getAsInt() != 3) {
            throw new Exception(new StringBuilder(512)
                    .append("parse low_fetch_limit rule failure in record(")
                    .append(recordNo).append(") of flowCtrlInfo value!").toString());
        }
        if (!jsonObject.has("rule")) {
            throw new Exception(new StringBuilder(512)
                    .append("FIELD rule is required in low_fetch_limit record(")
                    .append(recordNo).append(") of flowCtrlInfo value!").toString());
        }
        JsonArray ruleArray = jsonObject.get("rule").getAsJsonArray();
        if (ruleArray == null) {
            throw new Exception(new StringBuilder(512)
                    .append("emtpy rule define in low_fetch_limit record(")
                    .append(recordNo).append(") of flowCtrlInfo value!").toString());
        }
        if (ruleArray.size() > 1) {
            throw new Exception(new StringBuilder(512)
                    .append("only allow set one rule in low_fetch_limit record(")
                    .append(recordNo).append(") of flowCtrlInfo value!").toString());
        }
        int itemNo;
        ArrayList<FlowCtrlItem> flowCtrlItems = new ArrayList<>();
        // parse low_fetch_limit rule record
        for (int index = 0; index < ruleArray.size(); index++) {
            itemNo = index + 1;
            JsonObject ruleObject = ruleArray.get(index).getAsJsonObject();
            int normfreqInMs = 0;
            int filterFreqInMs = 0;
            int minDataFilterFreqInMs = 0;
            if (ruleObject.has("filterFreqInMs")
                    || ruleObject.has("minDataFilterFreqInMs")) {
                if (!ruleObject.has("filterFreqInMs")) {
                    throw new Exception(new StringBuilder(512)
                            .append("FIELD filterFreqInMs is required ")
                            .append("in low_fetch_limit item(").append(itemNo)
                            .append(").record(").append(recordNo)
                            .append(") of flowCtrlInfo value!").toString());
                }
                filterFreqInMs = ruleObject.get("filterFreqInMs").getAsInt();
                if (filterFreqInMs < 0 || filterFreqInMs > 300000) {
                    throw new Exception(new StringBuilder(512)
                            .append("the value of FIELD filterFreqInMs must in [0, 300000] ")
                            .append("in low_fetch_limit item(").append(itemNo)
                            .append(").record(").append(recordNo)
                            .append(") of flowCtrlInfo value!").toString());
                }
                if (!ruleObject.has("minDataFilterFreqInMs")) {
                    throw new Exception(new StringBuilder(512)
                            .append("FIELD minDataFilterFreqInMs is required ")
                            .append("in low_fetch_limit item(").append(itemNo)
                            .append(").record(").append(recordNo)
                            .append(") of flowCtrlInfo value!").toString());
                }
                minDataFilterFreqInMs = ruleObject.get("minDataFilterFreqInMs").getAsInt();
                if (minDataFilterFreqInMs < 0 || minDataFilterFreqInMs > 300000) {
                    throw new Exception(new StringBuilder(512)
                            .append("the value of FIELD minDataFilterFreqInMs must in [0, 300000] ")
                            .append("in low_fetch_limit item(").append(itemNo)
                            .append(").record(").append(recordNo)
                            .append(") of flowCtrlInfo value!").toString());
                }
                if (minDataFilterFreqInMs < filterFreqInMs) {
                    throw new Exception(new StringBuilder(512)
                            .append("the value of FIELD minDataFilterFreqInMs must be greater ")
                            .append("than or equal to the value of FIELD filterFreqInMs")
                            .append("in low_fetch_limit item(").append(itemNo)
                            .append(").record(").append(recordNo)
                            .append(") of flowCtrlInfo value!").toString());
                }
            }
            if (ruleObject.has("normFreqInMs")) {
                normfreqInMs = ruleObject.get("normFreqInMs").getAsInt();
                if (normfreqInMs < 0 || normfreqInMs > 300000) {
                    throw new Exception(new StringBuilder(512)
                            .append("the value of FIELD normFreqInMs must in [0, 300000] ")
                            .append("in low_fetch_limit item(").append(itemNo)
                            .append(").record(").append(recordNo)
                            .append(") of flowCtrlInfo value!").toString());
                }
            }
            flowCtrlItems.add(new FlowCtrlItem(typeVal,
                    normfreqInMs, filterFreqInMs, minDataFilterFreqInMs));
        }
        if (flowCtrlItems.isEmpty()) {
            throw new Exception(new StringBuilder(512)
                    .append("not found valid rule define in low_fetch_limit record(")
                    .append(recordNo).append(") of flowCtrlInfo value!").toString());
        }
        // sort rule set by the value of filterFreqInMs
        Collections.sort(flowCtrlItems, new Comparator<FlowCtrlItem>() {

            @Override
            public int compare(final FlowCtrlItem o1, final FlowCtrlItem o2) {
                if (o1.getFreqLtInMs() > o2.getFreqLtInMs()) {
                    return -1;
                } else if (o1.getFreqLtInMs() < o2.getFreqLtInMs()) {
                    return 1;
                } else {
                    return 0;
                }
            }
        });
        return flowCtrlItems;
    }
Parse low frequent fetch count

@param recordNo    record no
@param typeVal     type value
@param jsonObject  record json value
@return            parsed result
@throws Exception   Exception thrown
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corebase/policies/FlowCtrlRuleHandler.java