@Override
    public boolean updTopicDeployStatusInfo(BaseEntity opEntity, int brokerId,
            String topicName, TopicStatus topicStatus,
            StringBuilder strBuff, ProcessResult result) {
        TopicDeployEntity curEntity;
        TopicDeployEntity newEntity;
        String printPrefix = "[updTopicDeployConf], ";
        Integer topicLockId = null;
        Integer brokerLockId = null;
        // execute add or update operation
        try {
            // lock topicName meta-lock
            topicLockId = metaRowLock.getLock(null,
                    StringUtils.getBytesUtf8(topicName), true);
            try {
                // lock brokerId meta-lock
                brokerLockId = metaRowLock.getLock(null,
                        StringUtils.getBytesUtf8(String.valueOf(brokerId)), true);
                // check broker configure exist
                BrokerConfEntity brokerEntity =
                        brokerConfigMapper.getBrokerConfByBrokerId(brokerId);
                if (brokerEntity == null) {
                    result.setFailResult(DataOpErrCode.DERR_NOT_EXIST.getCode(),
                            strBuff.append("Not found broker configure by brokerId=")
                                    .append(brokerId)
                                    .append(", please create the broker's configure first!").toString());
                    strBuff.delete(0, strBuff.length());
                    return result.isSuccess();
                }
                // check topic deploy configure
                curEntity = topicDeployMapper.getTopicConf(brokerId, topicName);
                if (curEntity == null) {
                    result.setFailResult(DataOpErrCode.DERR_NOT_EXIST.getCode(),
                            strBuff.append("Not found topic deploy configure for brokerId-topicName(")
                                    .append(brokerId).append("-").append(topicName)
                                    .append(")!").toString());
                    strBuff.delete(0, strBuff.length());
                    return result.isSuccess();
                }
                if (curEntity.getTopicStatus() == topicStatus) {
                    result.setSuccResult(null);
                    return result.isSuccess();
                }
                // update record
                topicDeployMapper.updTopicDeployStatus(opEntity,
                        brokerId, topicName, topicStatus, strBuff, result);
                newEntity = topicDeployMapper.getTopicConfByeRecKey(curEntity.getRecordKey());
            } finally {
                if (brokerLockId != null) {
                    metaRowLock.releaseRowLock(brokerLockId);
                }
            }
        } catch (Throwable e) {
            return logExceptionInfo(e, printPrefix, strBuff, result);
        } finally {
            if (topicLockId != null) {
                metaRowLock.releaseRowLock(topicLockId);
            }
        }
        // print log to file
        if (result.isSuccess()) {
            strBuff.append(printPrefix).append(opEntity.getModifyUser())
                    .append(" updated topic deploy configure from ")
                    .append(curEntity).append(" to ").append(newEntity);
            logger.info(strBuff.toString());
            strBuff.delete(0, strBuff.length());
        }
        return result.isSuccess();
    }
Add or Update topic 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