private void processAndWaitingSave(ChannelHandlerContext ctx,
            ProxySdk.MessagePack packObject,
            List<ProxyEvent> events) throws Exception {
        ProxySdk.MessagePackHeader header = packObject.getHeader();
        InlongTcpSourceCallback callback = new InlongTcpSourceCallback(ctx, header);
        String inlongGroupId = header.getInlongGroupId();
        String inlongStreamId = header.getInlongStreamId();
        ProxyPackEvent packEvent = new ProxyPackEvent(inlongGroupId, inlongStreamId, events, callback);
        // put to channel
        try {
            source.getCachedChProcessor().processEvent(packEvent);
            events.forEach(event -> {
                source.addMetric(true, event.getBody().length, event);
                source.fileMetricIncSumStats(StatConstants.EVENT_MSG_V1_POST_SUCCESS);
            });
            boolean awaitResult = callback.getLatch().await(
                    CommonConfigHolder.getInstance().getMaxResAfterSaveTimeout(), TimeUnit.MILLISECONDS);
            if (!awaitResult) {
                if (!callback.getHasResponsed().getAndSet(true)) {
                    this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT, packObject);
                }
            }
        } catch (Throwable ex) {
            logger.error("Process Controller Event error can't write event to channel.", ex);
            events.forEach(event -> {
                source.addMetric(false, event.getBody().length, event);
                source.fileMetricIncSumStats(StatConstants.EVENT_MSG_V1_POST_DROPPED);
            });
            if (!callback.getHasResponsed().getAndSet(true)) {
                this.responsePackage(ctx, ProxySdk.ResultCode.ERR_REJECT, packObject);
            }
        }
    }
processAndWaitingSave
@param ctx
@param packObject
@param events
@throws Exception
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java