@Override
    public TubeMQResult addTopicsToBrokers(MasterEntry masterEntry, List<Integer> brokerIds,
            List<AddTopicReq> addTopicReqs) {
        TubeMQResult tubeResult = new TubeMQResult();
        AddTopicsResult addTopicsResult = new AddTopicsResult();

        if (CollectionUtils.isEmpty(addTopicReqs)) {
            return tubeResult;
        }
        addTopicReqs.forEach(addTopicReq -> {
            try {
                String brokerStr = StringUtils.join(brokerIds, ",");
                addTopicReq.setBrokerId(brokerStr);
                TubeMQResult result = addTopicToBrokers(addTopicReq, masterEntry);
                if (result.getErrCode() == TubeConst.SUCCESS_CODE) {
                    addTopicsResult.getSuccessTopics().add(addTopicReq.getTopicName());
                } else {
                    addTopicsResult.getFailTopics().add(addTopicReq.getTopicName());
                }
            } catch (Exception e) {
                log.error("add topic to brokers fail with exception", e);
                addTopicsResult.getFailTopics().add(addTopicReq.getTopicName());
            }
        });

        tubeResult.setData(addTopicsResult);
        return tubeResult;
    }
clone source broker to generate brokers with the same config and copy the topics in it.

@param req
@return

@throws Exception exception
inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/NodeServiceImpl.java