private void testRestartSource(Integer sourceId) {
        groupService.updateStatus(GROUP_ID, GroupStatus.CONFIG_OFFLINE_ING.getCode(), GLOBAL_OPERATOR);
        groupService.update(groupInfo.genRequest(), GLOBAL_OPERATOR);
        groupService.updateStatus(GROUP_ID, GroupStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode(), GLOBAL_OPERATOR);
        groupService.update(groupInfo.genRequest(), GLOBAL_OPERATOR);

        sourceService.updateStatus(GROUP_ID, null, SourceStatus.SOURCE_NORMAL.getCode(), GLOBAL_OPERATOR);

        GroupResourceProcessForm form = new GroupResourceProcessForm();
        form.setGroupInfo(groupInfo);
        form.setGroupOperateType(GroupOperateType.RESTART);
        WorkflowContext context = processService.start(ProcessName.RESTART_GROUP_PROCESS.name(), GLOBAL_OPERATOR, form);
        WorkflowResult result = WorkflowUtils.getResult(context);
        ProcessResponse response = result.getProcessInfo();
        Assertions.assertSame(response.getStatus(), ProcessStatus.COMPLETED);

        WorkflowProcess process = context.getProcess();
        WorkflowTask task = process.getTaskByName("RestartSource");
        Assertions.assertTrue(task instanceof ServiceTask);

        StreamSource streamSource = sourceService.get(sourceId);
        Assertions.assertSame(SourceStatus.forCode(streamSource.getStatus()), SourceStatus.TO_BE_ISSUED_RETRY);
    }
There will be concurrency problems in the overall operation,This method temporarily fails the test
inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/listener/StreamSourceListenerTest.java