@Override
    public ListenerResult listen(WorkflowContext context) throws WorkflowListenerException {
        StreamResourceProcessForm streamProcessForm = (StreamResourceProcessForm) context.getProcessForm();
        final String groupId = streamProcessForm.getInlongGroupId();
        InlongStreamInfo streamInfo = streamProcessForm.getStreamInfo();
        if (streamInfo == null) {
            String msg = "inlong stream cannot be null in StreamResourceProcessForm with groupId=" + groupId;
            log.error(msg);
            throw new WorkflowListenerException(msg);
        }

        // 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);
        }
        GroupStatus groupStatus = GroupStatus.forCode(groupInfo.getStatus());
        Preconditions.expectTrue(GroupStatus.CONFIG_FAILED != groupStatus,
                String.format("group status=%s not support start stream for groupId=%s", groupStatus, groupId));
        final String streamId = streamInfo.getInlongStreamId();
        // Read the current information
        streamProcessForm.setGroupInfo(groupInfo);
        streamProcessForm.setStreamInfo(streamService.get(groupId, streamId));

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

        QueueResourceOperator queueOperator = queueOperatorFactory.getInstance(groupInfo.getMqType());
        GroupOperateType operateType = streamProcessForm.getGroupOperateType();
        String operator = context.getOperator();
        switch (operateType) {
            case INIT:
                queueOperator.createQueueForStream(groupInfo, streamInfo, operator);
                break;
            case DELETE:
                queueOperator.deleteQueueForStream(groupInfo, streamInfo, operator);
                break;
            default:
                log.warn("unsupported operate={} for inlong group", operateType);
                break;
        }

        log.info("success to execute StreamQueueResourceListener for groupId={}, operateType={}", groupId, operateType);
        return ListenerResult.success("success");
    }
Create task listener for Pulsar Topic
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/StreamQueueResourceListener.java