@Override
    public ListenerResult listen(WorkflowContext context) throws WorkflowListenerException {
        StreamResourceProcessForm form = (StreamResourceProcessForm) context.getProcessForm();
        InlongStreamInfo streamInfo = form.getStreamInfo();
        final String groupId = streamInfo.getInlongGroupId();
        final String streamId = streamInfo.getInlongStreamId();
        // Read the current information
        InlongGroupInfo groupInfo = groupService.get(groupId);
        if (groupInfo == null) {
            String msg = "inlong group not found with groupId=" + groupId;
            LOGGER.error(msg);
            throw new WorkflowListenerException(msg);
        }
        form.setGroupInfo(groupInfo);
        form.setStreamInfo(streamService.get(groupId, streamId));
        groupInfo = form.getGroupInfo();
        streamInfo = form.getStreamInfo();
        LOGGER.info("begin to build sort config for groupId={}, streamId={}", groupId, streamId);

        GroupOperateType operateType = form.getGroupOperateType();
        if (operateType == GroupOperateType.SUSPEND || operateType == GroupOperateType.DELETE) {
            LOGGER.info("not build sort config for groupId={}, streamId={}, as the group operate type={}",
                    groupId, streamId, operateType);
            return ListenerResult.success();
        }

        GroupStatus groupStatus = GroupStatus.forCode(groupInfo.getStatus());
        Preconditions.expectTrue(GroupStatus.CONFIG_FAILED != groupStatus,
                String.format("group status=%s not support start stream for groupId=%s", groupStatus, groupId));
        List<StreamSink> streamSinks = streamInfo.getSinkList();
        if (CollectionUtils.isEmpty(streamSinks)) {
            LOGGER.warn("not build sort config for groupId={}, streamId={}, as not found any sinks", groupId, streamId);
            return ListenerResult.success();
        }

        try {

            List<String> sinkTypeList = streamSinks.stream().map(StreamSink::getSinkType).collect(Collectors.toList());
            List<SortConfigOperator> operatorList = operatorFactory.getInstance(sinkTypeList);
            for (SortConfigOperator operator : operatorList) {
                operator.buildConfig(groupInfo, streamInfo,
                        InlongConstants.SYNC_SEND.equals(groupInfo.getInlongGroupMode()));
            }

        } catch (Exception e) {
            String msg = String.format("failed to build sort config for groupId=%s, streamId=%s, ", groupId, streamId);
            LOGGER.error(msg + "streamInfo=" + streamInfo, e);
            throw new WorkflowListenerException(msg + e.getMessage());
        }

        LOGGER.info("success to build sort config for groupId={}, streamId={}", groupId, streamId);
        return ListenerResult.success();
    }
Event listener of build the Sort config for one inlong stream,
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/StreamSortConfigListener.java