private void processAndResponse(ChannelHandlerContext ctx,
            ProxySdk.MessagePack packObject,
            List<ProxyEvent> events) throws Exception {
        for (ProxyEvent event : events) {
            // get configured topic name
            String topic = configManager.getTopicName(event.getInlongGroupId(), event.getInlongStreamId());
            if (StringUtils.isEmpty(topic)) {
                source.fileMetricIncWithDetailStats(
                        StatConstants.EVENT_SOURCE_TOPIC_MISSING, event.getInlongGroupId());
                source.addMetric(false, event.getBody().length, event);
                this.responsePackage(ctx, ProxySdk.ResultCode.ERR_ID_ERROR, packObject);
                return;
            }
            event.setTopic(topic);
            // put to channel
            try {
                source.getCachedChProcessor().processEvent(event);
                source.addMetric(true, event.getBody().length, event);
                source.fileMetricIncSumStats(StatConstants.EVENT_MSG_V1_POST_SUCCESS);
            } catch (Throwable ex) {
                logger.error("Process Controller Event error can't write event to channel.", ex);
                source.addMetric(false, event.getBody().length, event);
                this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT, packObject);
                source.fileMetricIncSumStats(StatConstants.EVENT_MSG_V1_POST_DROPPED);
                return;
            }
        }
        this.responsePackage(ctx, ProxySdk.ResultCode.SUCCUSS, packObject);
    }
processAndResponse
@param ctx
@param packObject
@param events
@throws Exception
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java