@Transactional(rollbackOn = Exception.class)
    public void updateCreateTopicTaskStatus(long clusterId) {
        List<TopicTaskEntry> topicTasks = topicTaskRepository
                .findTopicTaskEntriesByClusterIdAndStatus(clusterId,
                        TaskStatusEnum.ADDING.getCode());
        TopicView topicView = topicService
                .requestTopicViewInfo(clusterId, null);
        if (topicView == null || topicView.getData() == null) {
            return;
        }
        List<TopicView.TopicViewInfo> topicViews = topicView.getData();
        Map<String, TopicView.TopicViewInfo> topicViewMap = topicViews.stream()
                .collect(Collectors.toMap(TopicView.TopicViewInfo::getTopicName, topicViewInfo -> topicViewInfo));
        MasterEntry masterNode = masterService.getMasterNode(clusterId);
        TubeHttpBrokerInfoList brokerInfoList = nodeService.requestBrokerStatus(masterNode);
        if (ValidateUtils.isNull(brokerInfoList)) {
            return;
        }
        updateTaskRepo(topicTasks, topicViewMap, brokerInfoList);
    }
Modify the status of creating a topic task

@param clusterId this is the cluster id
inlong-tubemq/tubemq-manager/src/main/java/org/apache/inlong/tubemq/manager/service/TaskServiceImpl.java