public SenderMessage fetchSenderMessage(String streamId, LinkedBlockingQueue<ProxyMessage> messageQueue) {
        int resultBatchSize = 0;
        List<byte[]> bodyList = new ArrayList<>();
        List<OffsetAckInfo> offsetList = 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);
                messageQueue.remove();
                break;
            }
            resultBatchSize += bodySize;
            // decrease queue size.
            bodyList.add(message.getBody());
            offsetList.add(message.getAckInfo());
        }
        // make sure result is not empty.
        long auditTime = 0;
        if (isRealTime) {
            auditTime = AgentUtils.getCurrentTime();
        } else {
            auditTime = dataTime;
        }
        if (!bodyList.isEmpty()) {
            SenderMessage senderMessage = new SenderMessage(taskId, instanceId, groupId, streamId, bodyList,
                    auditTime, extraMap, offsetList);
            return senderMessage;
        }
        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/file/ProxyMessageCache.java