public void addEvent(ProxyEvent event) {
        // parse
        String eventUid = event.getUid();
        long dispatchTime = event.getMsgTime() - event.getMsgTime() % MINUTE_MS;
        String dispatchKey = eventUid + "." + dispatchTime;
        // find dispatch profile
        PackProfile dispatchProfile = this.profileCache.get(dispatchKey);
        if (dispatchProfile == null) {
            dispatchProfile = new BatchPackProfile(eventUid, event.getInlongGroupId(),
                    event.getInlongStreamId(), dispatchTime);
            this.profileCache.put(dispatchKey, dispatchProfile);
        }
        // add event
        if (!dispatchProfile.addEvent(event, maxPackCount, maxPackSize)) {
            BatchPackProfile newDispatchProfile = new BatchPackProfile(eventUid, event.getInlongGroupId(),
                    event.getInlongStreamId(), dispatchTime);
            PackProfile oldDispatchProfile = this.profileCache.put(dispatchKey, newDispatchProfile);
            if (oldDispatchProfile != null) {
                this.mqZoneSink.acquireAndOfferDispatchedRecord(oldDispatchProfile);
            }
            this.outCounter.addAndGet(dispatchProfile.getCount());
            newDispatchProfile.addEvent(event, maxPackCount, maxPackSize);
        }
        this.inCounter.incrementAndGet();
    }
addEvent
@param event the event to add
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackManager.java