@Override
    public boolean addOrUpdTopicDeployInfo(boolean isAddOp, TopicDeployEntity entity,
            StringBuilder strBuff, ProcessResult result) {
        TopicDeployEntity curEntity;
        TopicDeployEntity newEntity;
        String printPrefix = "[addTopicDeployConf], ";
        Integer topicLockId = null;
        Integer brokerLockId = null;
        // add topic control configure
        BaseEntity opEntity = new BaseEntity("systemSelf", new Date());
        addTopicCtrlConfIfAbsent(opEntity, entity.getTopicName(), strBuff, result);
        // execute add or update operation
        try {
            // lock topicName meta-lock
            topicLockId = metaRowLock.getLock(null,
                    StringUtils.getBytesUtf8(entity.getTopicName()), true);
            try {
                // lock brokerId meta-lock
                brokerLockId = metaRowLock.getLock(null,
                        StringUtils.getBytesUtf8(String.valueOf(entity.getBrokerId())), true);
                // check broker configure exist
                BrokerConfEntity brokerEntity =
                        brokerConfigMapper.getBrokerConfByBrokerId(entity.getBrokerId());
                if (brokerEntity == null) {
                    result.setFailResult(DataOpErrCode.DERR_NOT_EXIST.getCode(),
                            strBuff.append("Not found broker configure by brokerId=")
                                    .append(entity.getBrokerId())
                                    .append(", please create the broker's configure first!").toString());
                    strBuff.delete(0, strBuff.length());
                    return result.isSuccess();
                }
                // check topic deploy configure
                curEntity = topicDeployMapper.getTopicConfByeRecKey(entity.getRecordKey());
                if (isAddOp) {
                    if (curEntity != null) {
                        if (curEntity.isValidTopicStatus()) {
                            result.setFailResult(DataOpErrCode.DERR_EXISTED.getCode(),
                                    strBuff.append("Existed record found for brokerId-topicName(")
                                            .append(curEntity.getRecordKey()).append(")!").toString());
                        } else {
                            result.setFailResult(DataOpErrCode.DERR_EXISTED.getCode(),
                                    strBuff.append("Softly deleted record found for brokerId-topicName(")
                                            .append(curEntity.getRecordKey())
                                            .append("), please resume or remove it first!").toString());
                        }
                        strBuff.delete(0, strBuff.length());
                        return result.isSuccess();
                    }
                    // add record
                    TopicPropGroup newProps =
                            getClusterDefSetting(false).getClsDefTopicProps().clone();
                    newProps.updModifyInfo(brokerEntity.getTopicProps());
                    newEntity = new TopicDeployEntity(entity,
                            entity.getBrokerId(), entity.getTopicName(), newProps);
                    newEntity.updModifyInfo(entity.getDataVerId(), entity.getTopicId(),
                            brokerEntity.getBrokerPort(), brokerEntity.getBrokerIp(),
                            entity.getTopicStatus(), entity.getTopicProps());
                    topicDeployMapper.addTopicDeployConf(newEntity, strBuff, result);
                } else {
                    printPrefix = "[updTopicDeployConf], ";
                    if (curEntity == null) {
                        result.setFailResult(DataOpErrCode.DERR_NOT_EXIST.getCode(),
                                strBuff.append("Not found topic deploy configure for brokerId-topicName(")
                                        .append(entity.getRecordKey()).append(")!").toString());
                        strBuff.delete(0, strBuff.length());
                        return result.isSuccess();
                    }
                    // update record
                    topicDeployMapper.updTopicDeployConf(entity, strBuff, result);
                }
                newEntity = topicDeployMapper.getTopicConfByeRecKey(entity.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()) {
            if (isAddOp) {
                strBuff.append(printPrefix).append(entity.getCreateUser())
                        .append(" added topic deploy configure: ").append(newEntity);
            } else {
                strBuff.append(printPrefix).append(entity.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