@Override
    public ListenerResult listen(WorkflowContext context) throws WorkflowListenerException {
        GroupResourceProcessForm form = (GroupResourceProcessForm) context.getProcessForm();
        String groupId = form.getInlongGroupId();
        LOGGER.info("begin to build sort config for groupId={}", groupId);

        GroupOperateType operateType = form.getGroupOperateType();
        if (operateType == GroupOperateType.SUSPEND || operateType == GroupOperateType.DELETE) {
            LOGGER.info("no need to build sort config for groupId={} as the operate type is {}", groupId, operateType);
            return ListenerResult.success();
        }
        // ensure the inlong group exists
        switch (operateType) {
            case INIT:
                groupService.updateStatus(groupId, GroupStatus.CONFIG_ING.getCode(), context.getOperator());
                break;
            case RESTART:
                groupService.updateStatus(groupId, GroupStatus.CONFIG_ONLINE_ING.getCode(), context.getOperator());
                break;
        }
        InlongGroupInfo groupInfo = groupService.get(groupId);
        if (groupInfo == null) {
            String msg = "inlong group not found with groupId=" + groupId;
            LOGGER.error(msg);
            throw new WorkflowListenerException(msg);
        }
        // Read the current information
        form.setGroupInfo(groupInfo);
        form.setStreamInfos(streamService.list(groupId));
        List<InlongStreamInfo> streamInfos = form.getStreamInfos();
        if (CollectionUtils.isEmpty(streamInfos)) {
            LOGGER.warn("no need to build sort config for groupId={}, as not found any stream", groupId);
            return ListenerResult.success();
        }

        int sinkCount = streamInfos.stream()
                .map(stream -> stream.getSinkList() == null ? 0 : stream.getSinkList().size())
                .reduce(0, Integer::sum);
        if (sinkCount == 0) {
            LOGGER.warn("not build sort config for groupId={}, as not found any sink", groupId);
            return ListenerResult.success();
        }

        try {
            for (InlongStreamInfo streamInfo : streamInfos) {
                List<StreamSink> sinkList = streamInfo.getSinkList();
                if (CollectionUtils.isEmpty(sinkList)) {
                    continue;
                }
                List<String> sinkTypeList = sinkList.stream().map(StreamSink::getSinkType).collect(Collectors.toList());
                List<SortConfigOperator> operatorList = operatorFactory.getInstance(sinkTypeList);
                for (SortConfigOperator operator : operatorList) {
                    operator.buildConfig(groupInfo, streamInfo,
                            InlongConstants.STANDARD_MODE.equals(groupInfo.getInlongGroupMode()));
                }
            }
        } catch (Exception e) {
            String msg = String.format("Failed to build sort config for group=%s, ", groupId);
            LOGGER.error("{} streamInfos={}", msg, streamInfos, e);
            throw new WorkflowListenerException(msg + e.getMessage());
        }

        LOGGER.info("success to build sort config for groupId={}", groupId);
        return ListenerResult.success();
    }
Event listener of build the Sort config,
such as update the form config, or build and push config to ZK, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java