@Override
    public ListenerResult listen(WorkflowContext context) throws WorkflowListenerException {
        ClusterResourceProcessForm clusterProcessForm = (ClusterResourceProcessForm) context.getProcessForm();
        log.info("begin to execute ClusterQueueResourceListener for clusterTag={},",
                clusterProcessForm.getInlongClusterTag());
        List<GroupFullInfo> groupFullInfoList = clusterProcessForm.getGroupFullInfoList();
        String operator = context.getOperator();
        for (GroupFullInfo groupFullInfo : groupFullInfoList) {
            InlongGroupInfo groupInfo = groupFullInfo.getGroupInfo();
            QueueResourceOperator queueOperator = queueOperatorFactory.getInstance(groupInfo.getMqType());
            queueOperator.createQueueForGroup(groupInfo, operator);
            for (InlongStreamInfo streamInfo : groupFullInfo.getStreamInfoList()) {
                queueOperator.createQueueForStream(groupInfo, streamInfo, operator);
            }
        }

        log.info("success to execute ClusterQueueResourceListener for clusterTag={}",
                clusterProcessForm.getInlongClusterTag());
        return ListenerResult.success("success");
    }
Create task listener for MQ Topic
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/ClusterQueueResourceListener.java