private boolean addOrUpdConsumeCtrlConf(boolean chkConsistent, boolean isAddOpOrOnlyAdd,
            GroupConsumeCtrlEntity entity, StringBuilder strBuff,
            ProcessResult result) {
        boolean addRecord = true;
        Integer topicLockId = null;
        Integer groupLockId = null;
        GroupConsumeCtrlEntity curEntity;
        GroupConsumeCtrlEntity newEntity;
        String printPrefix = "[addConsumeCtrlConf], ";
        // append topic control configure
        BaseEntity opEntity = new BaseEntity("systemSelf", new Date());
        addTopicCtrlConfIfAbsent(opEntity, entity.getTopicName(), strBuff, result);
        // append group control configure
        addGroupCtrlConfIfAbsent(opEntity, entity.getGroupName(), strBuff, result);
        // execute add or update operation
        try {
            // lock topicName meta-lock
            topicLockId = metaRowLock.getLock(null,
                    StringUtils.getBytesUtf8(entity.getTopicName()), true);
            try {
                // lock groupName meta-lock
                groupLockId = metaRowLock.getLock(null,
                        StringUtils.getBytesUtf8(entity.getGroupName()), true);
                // check consume control configure exist
                curEntity = consumeCtrlMapper.getGroupConsumeCtrlConfByRecKey(entity.getRecordKey());
                if (curEntity == null) {
                    if (chkConsistent && !isAddOpOrOnlyAdd) {
                        result.setFailResult(DataOpErrCode.DERR_NOT_EXIST.getCode(),
                                strBuff.append("Not found consume control for groupName-topicName(")
                                        .append(entity.getRecordKey()).append(")!").toString());
                        strBuff.delete(0, strBuff.length());
                        return result.isSuccess();
                    }
                    entity.fillEmptyValues();
                    consumeCtrlMapper.addGroupConsumeCtrlConf(entity, strBuff, result);
                } else {
                    if (isAddOpOrOnlyAdd) {
                        if (chkConsistent) {
                            result.setFailResult(DataOpErrCode.DERR_EXISTED.getCode(),
                                    strBuff.append("Existed record found for groupName-topicName(")
                                            .append(entity.getRecordKey()).append(")!").toString());
                            strBuff.delete(0, strBuff.length());
                        } else {
                            result.setSuccResult(null);
                        }
                        return result.isSuccess();
                    }
                    addRecord = false;
                    printPrefix = "[updConsumeCtrlConf], ";
                    consumeCtrlMapper.updGroupConsumeCtrlConf(entity, strBuff, result);
                }
                newEntity = consumeCtrlMapper.getGroupConsumeCtrlConfByRecKey(entity.getRecordKey());
            } finally {
                if (groupLockId != null) {
                    metaRowLock.releaseRowLock(groupLockId);
                }
            }
        } catch (Throwable e) {
            return logExceptionInfo(e, printPrefix, strBuff, result);
        } finally {
            if (topicLockId != null) {
                metaRowLock.releaseRowLock(topicLockId);
            }
        }
        // print log to file
        if (result.isSuccess()) {
            if (addRecord) {
                strBuff.append(printPrefix).append(entity.getCreateUser())
                        .append(" added consume control configure: ").append(newEntity);
            } else {
                strBuff.append(printPrefix).append(entity.getModifyUser())
                        .append(" updated consume control configure from ")
                        .append(curEntity).append(" to ").append(newEntity);
            }
            logger.info(strBuff.toString());
            strBuff.delete(0, strBuff.length());
        }
        return result.isSuccess();
    }
Add or Update consume control configure info

@param chkConsistent     whether order operation condition
@param isAddOpOrOnlyAdd  the operation type,
@param entity            the entity need to operation
@param strBuff           the string buffer
@param result            the process result return
@return true if success otherwise false
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/master/metamanage/metastore/impl/AbsMetaConfigMapperImpl.java