@Override
    public ListenerResult listen(WorkflowContext context) {
        StreamResourceProcessForm form = (StreamResourceProcessForm) context.getProcessForm();
        InlongStreamInfo streamInfo = form.getStreamInfo();
        final String groupId = streamInfo.getInlongGroupId();
        final String streamId = streamInfo.getInlongStreamId();
        log.info("begin to create sink resource for groupId={}, streamId={}", groupId, streamId);
        InlongGroupInfo groupInfo = groupService.get(groupId);
        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<SinkInfo> sinkInfos = sinkEntityMapper.selectAllConfig(groupId, Lists.newArrayList(streamId));
        List<SinkInfo> needCreateResources = sinkInfos.stream()
                .filter(sinkInfo -> InlongConstants.ENABLE_CREATE_RESOURCE.equals(sinkInfo.getEnableCreateResource()))
                .collect(Collectors.toList());

        if (CollectionUtils.isEmpty(needCreateResources)) {
            log.info("sink resources have been created for group [" + groupId + "] and stream [" + streamId + "]");
            return ListenerResult.success();
        }

        for (SinkInfo sinkInfo : needCreateResources) {
            SinkResourceOperator resourceOperator = resourceOperatorFactory.getInstance(sinkInfo.getSinkType());
            resourceOperator.createSinkResource(sinkInfo);
        }
        log.info("success to create sink resources for group [" + groupId + "] and stream [" + streamId + "]");
        return ListenerResult.success();
    }
Event listener of operate sink resources for one inlong stream,
such as create or update Hive table, Kafka topics, ES indices, etc.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sink/StreamSinkResourceListener.java