public boolean startProcess(String groupId, String streamId, String operator, boolean sync) {
        log.info("begin to start stream process for groupId={} streamId={}", groupId, streamId);

        InlongGroupInfo groupInfo = groupService.get(groupId);
        Preconditions.expectNotNull(groupInfo, ErrorCodeEnum.GROUP_NOT_FOUND.getMessage());
        GroupStatus groupStatus = GroupStatus.forCode(groupInfo.getStatus());
        if (groupStatus != GroupStatus.CONFIG_SUCCESSFUL) {
            throw new BusinessException(String.format("group status=%s not support start stream"
                    + " for groupId=%s", groupStatus, groupId));
        }

        InlongStreamInfo streamInfo = streamService.get(groupId, streamId);
        Preconditions.expectNotNull(streamInfo, ErrorCodeEnum.STREAM_NOT_FOUND.getMessage());
        StreamStatus status = StreamStatus.forCode(streamInfo.getStatus());
        if (status == StreamStatus.CONFIG_ING) {
            log.warn("stream status={}, not need restart for groupId={} streamId={}", status, groupId, streamId);
            return true;
        }

        if (StreamStatus.notAllowedUpdate(status)) {
            String errMsg = String.format("stream status=%s not support start stream for groupId=%s streamId=%s",
                    status, groupId, streamId);
            log.error(errMsg);
            throw new BusinessException(errMsg);
        }

        StreamResourceProcessForm processForm = StreamResourceProcessForm.getProcessForm(groupInfo,
                streamInfo, GroupOperateType.INIT);
        ProcessName processName = ProcessName.CREATE_STREAM_RESOURCE;
        if (sync) {
            WorkflowResult workflowResult = workflowService.start(processName, operator, processForm);
            ProcessStatus processStatus = workflowResult.getProcessInfo().getStatus();
            return processStatus == ProcessStatus.COMPLETED;
        } else {
            UserInfo userInfo = LoginUserUtils.getLoginUser();
            EXECUTOR_SERVICE.execute(
                    () -> workflowService.startAsync(processName, userInfo, processForm));
            return true;
        }
    }
Create stream in synchronous/asynchronous way.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamProcessService.java