@Override
    public ListenerResult listen(WorkflowContext context) throws WorkflowListenerException {
        GroupResourceProcessForm groupProcessForm = (GroupResourceProcessForm) context.getProcessForm();
        final String groupId = groupProcessForm.getInlongGroupId();
        // ensure the inlong group exists
        InlongGroupInfo groupInfo = groupService.get(groupId);
        if (groupInfo == null) {
            String msg = "inlong group not found with groupId=" + groupId;
            log.error(msg);
            throw new WorkflowListenerException(msg);
        }
        // Read the current information
        groupProcessForm.setGroupInfo(groupInfo);
        groupProcessForm.setStreamInfos(streamService.list(groupId));

        String operator = context.getOperator();
        GroupOperateType operateType = groupProcessForm.getGroupOperateType();
        if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())
                || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
            log.warn("skip to execute QueueResourceListener as sync mode {} (1 for realtime sync, 2 for offline sync) "
                    + "for groupId={}", groupInfo.getInlongGroupMode(), groupId);
            if (GroupOperateType.INIT.equals(operateType)) {
                this.createQueueForStreams(groupInfo, groupProcessForm.getStreamInfos(), operator);
            }
            return ListenerResult.success("skip - disable create mq resource for sync mode");
        }

        if (InlongConstants.DISABLE_CREATE_RESOURCE.equals(groupInfo.getEnableCreateResource())) {
            log.warn("skip to execute QueueResourceListener as disable create resource for groupId={}", groupId);
            return ListenerResult.success("skip - disable create resource");
        }

        QueueResourceOperator queueOperator = queueOperatorFactory.getInstance(groupInfo.getMqType());
        switch (operateType) {
            case INIT:
                groupService.updateStatus(groupId, GroupStatus.CONFIG_ING.getCode(), operator);
                // create queue resource for inlong group
                queueOperator.createQueueForGroup(groupInfo, operator);
                // create queue resource for all inlong streams under the inlong group
                this.createQueueForStreams(groupInfo, groupProcessForm.getStreamInfos(), operator);
                break;
            case DELETE:
                groupService.updateStatus(groupId, GroupStatus.CONFIG_DELETING.getCode(), operator);
                queueOperator.deleteQueueForGroup(groupInfo, operator);
                break;
            default:
                log.warn("unsupported operate={} for inlong group", operateType);
                break;
        }

        log.info("success to execute QueueResourceListener for groupId={}, operateType={}", groupId, operateType);
        return ListenerResult.success("success");
    }
Create message queue resources,
such as Pulsar Topic and Subscription, TubeMQ Topic and ConsumerGroup, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java