public BatchProxyMessage fetchBatch() {
        // if queue is nearly full or package size is satisfied or timeout
        long currentTime = System.currentTimeMillis();
        if (queueSize.get() > maxPackSize || queueIsFull()
                || currentTime - currentCacheTime > cacheTimeout) {
            // refresh cache time.
            currentCacheTime = currentTime;
            long resultBatchSize = 0;
            List<byte[]> result = new ArrayList<>();
            while (!messageQueue.isEmpty()) {
                // pre check message size
                ProxyMessage peekMessage = messageQueue.peek();
                int peekMessageLength = peekMessage.getBody().length;
                if (resultBatchSize + peekMessageLength > maxPackSize) {
                    break;
                }
                ProxyMessage message = messageQueue.remove();
                int bodySize = message.getBody().length;
                if (peekMessageLength > maxPackSize) {
                    LOGGER.warn("message size is {}, greater than max pack size {}, drop it!",
                            peekMessage.getBody().length, maxPackSize);
                    queueSize.addAndGet(-bodySize);
                    messageQueue.remove();
                    break;
                }
                resultBatchSize += bodySize;
                // decrease queue size.
                queueSize.addAndGet(-bodySize);
                result.add(message.getBody());
            }
            // make sure result is not empty.
            if (!result.isEmpty()) {
                return new BatchProxyMessage(jobId, groupId, streamId, result, AgentUtils.getCurrentTime(), extraMap);
            }
        }
        return null;
    }
Fetch batch of proxy message, timeout message or max number of list satisfied.

@return map of message list, key is stream id for the batch; return null if there are no valid messages.
inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/message/PackProxyMessage.java