private void parseMixAttr() throws IOException {
        attr2Rawdata = new LinkedHashMap<String, DataByteBuffer>(
                this.msgcnt * 10 / 7);

        for (int i = 0; i < attrcnt; i++) {
            ByteBuffer bodyBuffer;
            String commonAttr = parsedInput.readUTF();
            int len = parsedInput.readInt();
            int compress = parsedInput.readByte();
            int pos = parsedInput.getPosition();

            if (compress == 1) {
                byte[] uncompressdata = new byte[Snappy.uncompressedLength(
                        parsedInput.getData(), pos, len - 1)];
                int msgLen = Snappy.uncompress(parsedInput.getData(), pos, len - 1,
                        uncompressdata, 0);
                bodyBuffer = ByteBuffer.wrap(uncompressdata, 0, msgLen);
            } else {
                bodyBuffer = ByteBuffer.wrap(parsedInput.getData(), pos, len - 1);
            }
            parsedInput.skip(len - 1);

            while (bodyBuffer.remaining() > 0) {
                // total message length = (data length + attributes length) * N
                int singleTotalLen = bodyBuffer.getInt();
                if (singleTotalLen > bodyBuffer.remaining()) {
                    return;
                }

                while (singleTotalLen > 0) {
                    // single data length
                    int msgItemLen = bodyBuffer.getInt();
                    if (msgItemLen <= 0 || msgItemLen > singleTotalLen) {
                        return;
                    }

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

                    // single attribute length
                    int singleAttrLen = bodyBuffer.getInt();
                    if (singleAttrLen <= 0 || singleAttrLen > singleTotalLen) {
                        return;
                    }
                    byte[] attrBuf = new byte[singleAttrLen];
                    bodyBuffer.get(attrBuf, 0, singleAttrLen);
                    String finalAttr = commonAttr + "&" + new String(attrBuf);

                    DataByteBuffer inputBuffer = attr2Rawdata.get(finalAttr);
                    if (inputBuffer == null) {
                        inputBuffer = new DataByteBuffer(0,
                                new DataOutputBuffer(msgItemLen + 4 + 1));
                        attr2Rawdata.put(finalAttr, inputBuffer);
                        inputBuffer.inoutBuffer.write(record, 0, msgItemLen + 4 + 1);
                    } else {
                        inputBuffer.inoutBuffer.write(record, 1, msgItemLen + 4);
                    }
                    singleTotalLen = singleTotalLen - msgItemLen - 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