private void createQueueForStreams(InlongGroupInfo groupInfo, List<InlongStreamInfo> streamInfos, String operator) {
        String groupId = groupInfo.getInlongGroupId();
        log.info("success to start stream process for groupId={}", groupId);

        for (InlongStreamInfo stream : streamInfos) {
            StreamResourceProcessForm form = StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
            String streamId = stream.getInlongStreamId();
            final String errMsg = "failed to start stream process for groupId=" + groupId + " streamId=" + streamId;
            UserInfo userInfo = LoginUserUtils.getLoginUser();
            CompletableFuture<WorkflowResult> future = CompletableFuture
                    .supplyAsync(() -> workflowService.startAsync(CREATE_STREAM_RESOURCE, userInfo, form),
                            EXECUTOR_SERVICE)
                    .whenComplete((result, ex) -> {
                        if (ex != null) {
                            log.error(errMsg + ": " + ex.getMessage());
                            throw new WorkflowListenerException(errMsg, ex);
                        } else {
                            List<TaskResponse> tasks = result.getNewTasks();
                            if (TaskStatus.FAILED == tasks.get(tasks.size() - 1).getStatus()) {
                                log.error(errMsg);
                                throw new WorkflowListenerException(errMsg);
                            }
                        }
                    });
            try {
                // wait for the current process complete before starting the next stream,
                // otherwise, an exception is thrown and the next stream process will not be started.
                future.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
            } catch (Exception e) {
                String msg = "failed to execute stream process in asynchronously ";
                log.error(msg, e);
                throw new WorkflowListenerException(msg + ": " + e.getMessage());
            }
        }

        log.info("success to start stream process for groupId={}", groupId);
    }
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