public boolean deleteProcess(String groupId, String streamId, String operator, boolean sync) {
        log.debug("begin to delete stream process for groupId={} streamId={}", groupId, streamId);

        InlongGroupInfo groupInfo = groupService.get(groupId);
        if (groupInfo == null) {
            throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND,
                    ErrorCodeEnum.GROUP_NOT_FOUND.getMessage() + " : " + groupId);
        }

        GroupStatus groupStatus = GroupStatus.forCode(groupInfo.getStatus());
        if (GroupStatus.notAllowedTransition(groupStatus, GroupStatus.CONFIG_DELETING)) {
            throw new BusinessException(ErrorCodeEnum.GROUP_DELETE_NOT_ALLOWED,
                    String.format("group status=%s not support delete 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.DELETED || status == StreamStatus.DELETING) {
            log.debug("groupId={}, streamId={} is already in {}", groupId, streamId, status);
            return true;
        }

        if (StreamStatus.notAllowedDelete(status)) {
            throw new BusinessException(ErrorCodeEnum.STREAM_OPT_NOT_ALLOWED,
                    String.format("stream status=%s not support delete stream for groupId=%s streamId=%s", status,
                            groupId, streamId));
        }

        StreamResourceProcessForm processForm =
                StreamResourceProcessForm.getProcessForm(groupInfo, streamInfo, GroupOperateType.DELETE);
        ProcessName processName = ProcessName.DELETE_STREAM_RESOURCE;
        if (sync) {
            WorkflowResult workflowResult = workflowService.start(processName, operator, processForm);
            ProcessStatus processStatus = workflowResult.getProcessInfo().getStatus();
            if (processStatus == ProcessStatus.COMPLETED) {
                return streamService.delete(groupId, streamId, operator);
            } else {
                return false;
            }
        } else {
            UserInfo userInfo = LoginUserUtils.getLoginUser();
            EXECUTOR_SERVICE.execute(() -> {
                LoginUserUtils.setUserLoginInfo(userInfo);
                WorkflowResult workflowResult = workflowService.start(processName, operator, processForm);
                ProcessStatus processStatus = workflowResult.getProcessInfo().getStatus();
                if (processStatus == ProcessStatus.COMPLETED) {
                    streamService.delete(groupId, streamId, operator);
                }
                LoginUserUtils.removeUserLoginInfo();
            });
            return true;
        }
    }
Restart stream in synchronous/asynchronous way.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamProcessService.java