private void parseBinMsg() throws IOException {
        Map<String, String> commonAttrMap = new HashMap<String, String>();

        int totalLen = parsedBinInput.getInt(BIN_MSG_TOTALLEN_OFFSET);
        final int msgtype = parsedBinInput.get(BIN_MSG_MSGTYPE_OFFSET);
        int groupIdNum = parsedBinInput.getShort(BIN_MSG_GROUPID_OFFSET);
        int streamIdNum = parsedBinInput.getShort(BIN_MSG_STREAMID_OFFSET);
        int bodyLen = parsedBinInput.getInt(BIN_MSG_BODYLEN_OFFSET);
        long dataTime = parsedBinInput.getInt(BIN_MSG_DATATIME_OFFSET);
        final int extField = parsedBinInput.getShort(BIN_MSG_EXTFIELD_OFFSET);
        int attrLen = parsedBinInput.getShort(BIN_MSG_BODY_OFFSET + bodyLen);
        int msgMagic = (parsedBinInput.getShort(BIN_MSG_BODY_OFFSET
                + bodyLen + BIN_MSG_ATTRLEN_SIZE + attrLen) & 0xFFFF);
        dataTime = dataTime * 1000;

        // read common attributes
        if (attrLen != 0) {
            byte[] attr = new byte[attrLen];
            parsedBinInput.position(BIN_MSG_BODY_OFFSET + bodyLen + BIN_MSG_ATTRLEN_SIZE);
            parsedBinInput.get(attr);
            String strAttr = new String(attr);

            commonAttrMap = new HashMap<String, String>(MAP_SPLITTER.split(strAttr));
        }

        commonAttrMap.put(AttributeConstants.DATA_TIME, String.valueOf(dataTime));

        // unzip data
        ByteBuffer bodyBuffer;
        byte[] body = new byte[bodyLen + 1];
        parsedBinInput.position(BIN_MSG_BODY_OFFSET);
        parsedBinInput.get(body, 1, bodyLen);
        int zipType = (msgtype & 0xE0) >> 5;
        switch (zipType) {
            case (BIN_MSG_SNAPPY_TYPE):
                byte[] uncompressdata =
                        new byte[Snappy.uncompressedLength(body, 1, body.length - 1) + 1];
                // uncompress flag
                uncompressdata[0] = 0;
                int msgLen = Snappy.uncompress(body, 1, body.length - 1,
                        uncompressdata, 1);
                bodyBuffer = ByteBuffer.wrap(uncompressdata, 0, msgLen + 1);
                break;

            case (BIN_MSG_NO_ZIP):
            default:
                // set uncompress flag
                body[0] = 0;
                bodyBuffer = ByteBuffer.wrap(body, 0, body.length);
                break;
        }

        // number groupId/streamId
        boolean isUseNumGroupId = ((extField & 0x4) == 0x0);
        if (isUseNumGroupId) {
            commonAttrMap.put(AttributeConstants.GROUP_ID, String.valueOf(groupIdNum));
            commonAttrMap.put(AttributeConstants.STREAM_ID, String.valueOf(streamIdNum));
        }

        boolean hasOtherAttr = ((extField & 0x1) == 0x1);
        commonAttrMap.put(AttributeConstants.MESSAGE_COUNT, String.valueOf(this.msgcnt));
        // with private attributes,
        // need to splice private attributes + public attributes
        if (!hasOtherAttr) {
            // general attributes and data map
            attr2Rawdata = new LinkedHashMap<String, DataByteBuffer>();
            attr2Rawdata.put(MAP_JOINER.join(commonAttrMap),
                    new DataByteBuffer(0, bodyBuffer));
        } else {
            attr2Rawdata = new LinkedHashMap<String, DataByteBuffer>(
                    this.msgcnt * 10 / 7);
            Map<String, String> finalAttrMap = commonAttrMap;

            // skip compress flag
            bodyBuffer.get();
            int bodyBufLen = bodyBuffer.capacity() - 1;
            while (bodyBufLen > 0) {
                // get single message length
                int singleMsgLen = bodyBuffer.getInt();
                if (singleMsgLen <= 0 || singleMsgLen > bodyBufLen) {
                    return;
                }

                byte[] record = new byte[1 + 4 + singleMsgLen];
                record[0] = 0;
                record[1] = (byte) ((singleMsgLen >> 24) & 0xFF);
                record[2] = (byte) ((singleMsgLen >> 16) & 0xFF);
                record[3] = (byte) ((singleMsgLen >> 8) & 0xFF);
                record[4] = (byte) (singleMsgLen & 0xFF);
                bodyBuffer.get(record, 1 + 4, singleMsgLen);

                // get single attribute length
                int singleAttrLen = bodyBuffer.getInt();
                if (singleAttrLen <= 0 || singleAttrLen > bodyBufLen) {
                    return;
                }
                byte[] attrBuf = new byte[singleAttrLen];
                bodyBuffer.get(attrBuf, 0, singleAttrLen);
                String attrBufStr = new String(attrBuf);

                finalAttrMap = new HashMap<String, String>(MAP_SPLITTER.split(attrBufStr));
                finalAttrMap.putAll(commonAttrMap);

                DataByteBuffer inputBuffer = attr2Rawdata.get(MAP_JOINER.join(finalAttrMap));
                if (inputBuffer == null) {
                    inputBuffer = new DataByteBuffer(0,
                            new DataOutputBuffer(singleMsgLen + 4 + 1));
                    attr2Rawdata.put(MAP_JOINER.join(finalAttrMap), inputBuffer);
                    inputBuffer.inoutBuffer.write(record, 0, singleMsgLen + 4 + 1);
                } else {
                    inputBuffer.inoutBuffer.write(record, 1, singleMsgLen + 4);
                }

                bodyBufLen = bodyBufLen - singleMsgLen - singleAttrLen - 8;
            }

            // sync data
            for (String attr : attr2Rawdata.keySet()) {
                DataByteBuffer data = attr2Rawdata.get(attr);
                data.syncByteBuffer();
            }
        }
    }
add msg
@param attr
@param data
@return
inlong-common/src/main/java/org/apache/inlong/common/msg/InLongMsg.java