public void addPackEvent(ProxyPackEvent packEvent) {
        String eventUid = packEvent.getUid();
        long dispatchTime = packEvent.getMsgTime() - packEvent.getMsgTime() % MINUTE_MS;
        BatchPackProfile dispatchProfile = new BatchPackProfile(eventUid, packEvent.getInlongGroupId(),
                packEvent.getInlongStreamId(), dispatchTime);
        // callback
        BatchPackProfileCallback callback = new BatchPackProfileCallback(packEvent.getEvents().size(),
                packEvent.getCallback());
        dispatchProfile.setCallback(callback);
        // offer queue
        for (ProxyEvent event : packEvent.getEvents()) {
            if (!dispatchProfile.addEvent(event, maxPackCount, maxPackSize)) {
                // dispatch profile is full
                this.outCounter.addAndGet(dispatchProfile.getCount());
                this.mqZoneSink.acquireAndOfferDispatchedRecord(dispatchProfile);
                dispatchProfile = new BatchPackProfile(eventUid, event.getInlongGroupId(), event.getInlongStreamId(),
                        dispatchTime);
                dispatchProfile.setCallback(callback);
                dispatchProfile.addEvent(event, maxPackCount, maxPackSize);
            }
            this.inCounter.incrementAndGet();
        }
        // last dispatch profile
        if (dispatchProfile.getEvents().size() > 0) {
            this.outCounter.addAndGet(dispatchProfile.getCount());
            this.mqZoneSink.acquireAndOfferDispatchedRecord(dispatchProfile);
        }
    }
addPackEvent
@param packEvent  the event need to add
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackManager.java