public static ClientBroker.TransferedMessage getTransferMsg(ByteBuffer dataBuffer, int dataTotalSize,
            HashMap<String, TrafficInfo> countMap,
            String statisKeyBase,
            StringBuilder sBuilder) {
        if (dataBuffer.array().length < dataTotalSize) {
            return null;
        }
        final int msgLen =
                dataBuffer.getInt(DataStoreUtils.STORE_HEADER_POS_LENGTH);
        final int msgToken =
                dataBuffer.getInt(DataStoreUtils.STORE_HEADER_POS_DATATYPE);
        final int checkSum =
                dataBuffer.getInt(DataStoreUtils.STORE_HEADER_POS_CHECKSUM);
        int payLoadLen = msgLen - DataStoreUtils.STORE_DATA_PREFX_LEN;
        int payLoadOffset = DataStoreUtils.STORE_DATA_HEADER_LEN;
        if ((msgToken != DataStoreUtils.STORE_DATA_TOKER_BEGIN_VALUE)
                || (payLoadLen <= 0)
                || (payLoadLen > dataTotalSize - DataStoreUtils.STORE_DATA_HEADER_LEN)) {
            return null;
        }
        final long msgId = dataBuffer.getLong(DataStoreUtils.STORE_HEADER_POS_MSGID);
        final int flag = dataBuffer.getInt(DataStoreUtils.STORE_HEADER_POS_MSGFLAG);
        final int payLoadLen2 = payLoadLen;
        final byte[] payLoadData = new byte[payLoadLen];
        System.arraycopy(dataBuffer.array(), payLoadOffset, payLoadData, 0, payLoadLen);
        ClientBroker.TransferedMessage.Builder dataBuilder =
                ClientBroker.TransferedMessage.newBuilder();
        dataBuilder.setMessageId(msgId);
        dataBuilder.setCheckSum(checkSum);
        dataBuilder.setFlag(flag);
        dataBuilder.setPayLoadData(ByteString.copyFrom(payLoadData));
        // get statistic data
        int attrLen = 0;
        String attribute = null;
        if (MessageFlagUtils.hasAttribute(flag)) {
            if (payLoadLen < 4) {
                return null;
            }
            attrLen = dataBuffer.getInt(DataStoreUtils.STORE_DATA_HEADER_LEN);
            payLoadOffset += 4;
            payLoadLen -= 4;
            if (attrLen > payLoadLen) {
                return null;
            }
            if (attrLen > 0) {
                final byte[] attrData = new byte[attrLen];
                System.arraycopy(dataBuffer.array(), payLoadOffset, attrData, 0, attrLen);
                try {
                    attribute = new String(attrData, TBaseConstants.META_DEFAULT_CHARSET_NAME);
                } catch (final UnsupportedEncodingException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        String messageTime = "";
        if (TStringUtils.isNotBlank(attribute)) {
            if (attribute.contains(TokenConstants.TOKEN_MSG_TIME)) {

                String[] strAttrs = attribute.split(TokenConstants.ARRAY_SEP);
                for (String strAttrItem : strAttrs) {
                    if (TStringUtils.isNotBlank(strAttrItem)) {
                        if (strAttrItem.contains(TokenConstants.TOKEN_MSG_TIME)) {
                            String[] strItems = strAttrItem.split(TokenConstants.EQ);
                            if (strItems.length > 1) {
                                messageTime = strItems[1];
                            }
                        }
                    }
                }
            }
        }
        String baseKey = sBuilder.append(statisKeyBase)
                .append("#").append(messageTime).toString();
        sBuilder.delete(0, sBuilder.length());
        TrafficInfo getCount = countMap.get(baseKey);
        if (getCount == null) {
            countMap.put(baseKey, new TrafficInfo(1L, payLoadLen2));
        } else {
            getCount.addMsgCntAndSize(1L, payLoadLen2);
        }
        ClientBroker.TransferedMessage transferedMessage = dataBuilder.build();
        dataBuilder.clear();
        return transferedMessage;
    }
Convert inner message to protobuf format, then reply to client.

@param dataBuffer      the raw stored data
@param dataTotalSize   the data size
@param countMap        the statistics map
@param statisKeyBase   the statistics key prefix
@param sBuilder        the string buffer
@return                the converted messages
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/utils/DataStoreUtils.java