public void testCreateSortConfigInUpdateWorkflow() {
        InlongGroupInfo groupInfo = createInlongGroup("test20", MQType.PULSAR);
        groupInfo.setEnableZookeeper(InlongConstants.ENABLE_ZK);
        groupInfo.setEnableCreateResource(InlongConstants.ENABLE_CREATE_RESOURCE);
        groupService.updateStatus(GROUP_ID, GroupStatus.CONFIG_SUCCESSFUL.getCode(), OPERATOR);
        groupService.update(groupInfo.genRequest(), OPERATOR);

        InlongStreamInfo streamInfo = createStreamInfo(groupInfo, "test_stream_info");
        createHiveSink(streamInfo);
        createKafkaSource(streamInfo);
        GroupResourceProcessForm form = new GroupResourceProcessForm();
        form.setGroupInfo(groupInfo);
        form.setGroupOperateType(GroupOperateType.SUSPEND);
        taskListenerFactory.acceptPlugin(new MockPlugin());

        WorkflowContext context = processService.start(ProcessName.SUSPEND_GROUP_PROCESS.name(), applicant, form);
        WorkflowResult result = WorkflowUtils.getResult(context);
        ProcessResponse response = result.getProcessInfo();
        Assertions.assertSame(response.getStatus(), ProcessStatus.COMPLETED);
        WorkflowProcess process = context.getProcess();
        WorkflowTask task = process.getTaskByName("StopSort");
        Assertions.assertTrue(task instanceof ServiceTask);
        Assertions.assertEquals(2, task.getNameToListenerMap().size());
        List<TaskEventListener> listeners = Lists.newArrayList(task.getNameToListenerMap().values());
        Assertions.assertEquals(2, listeners.size());
        ProcessForm currentProcessForm = context.getProcessForm();
        InlongGroupInfo curGroupRequest = ((GroupResourceProcessForm) currentProcessForm).getGroupInfo();
        Assertions.assertEquals(1, curGroupRequest.getExtList().size());
    }
Creat kafka source info by inlong stream info.
inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/sort/DisableZkForSortTest.java