private ByteBuffer binBuild(long createtime) {
        try {
            this.createtime = createtime;
            DataOutputBuffer out = new DataOutputBuffer(capacity);

            writeMagic(out);

            int msgType = getBinMsgtype(binMsgBuffer);
            int compressType = ((msgType & 0xE0) >> 5);
            if ((compressType == 0) && (compress)) {

                binMsgBuffer.position(BIN_MSG_BODYLEN_OFFSET);
                // copy body data
                int bodyLen = binMsgBuffer.getInt();
                byte[] body = new byte[bodyLen];
                binMsgBuffer.get(body, 0, bodyLen);

                // copy attributes
                int attrLen =
                        binMsgBuffer.getShort(BIN_MSG_BODY_OFFSET + bodyLen);
                byte[] attr =
                        new byte[BIN_MSG_ATTRLEN_SIZE + attrLen + BIN_MSG_MAGIC_SIZE];
                binMsgBuffer.get(attr, 0, attr.length);

                int guessLen = Snappy.maxCompressedLength(bodyLen);
                byte[] tmpData = new byte[guessLen];
                int realLen = Snappy.compress(body, 0,
                        body.length, tmpData, 0);

                int totalDataLen = binMsgBuffer.getInt(BIN_MSG_TOTALLEN_OFFSET);
                ByteBuffer dataBuf = ByteBuffer.allocate(
                        totalDataLen + BIN_MSG_TOTALLEN_SIZE - body.length + realLen);

                // copy headers
                dataBuf.put(binMsgBuffer.array(), 0, BIN_MSG_BODYLEN_OFFSET);
                // set compress flag
                dataBuf.put(BIN_MSG_MSGTYPE_OFFSET, (byte) (msgType | BIN_MSG_SET_SNAPPY));
                dataBuf.putInt(BIN_MSG_TOTALLEN_OFFSET,
                        realLen + attrLen + BIN_MSG_FORMAT_SIZE - 4);
                // set data length
                dataBuf.putInt(BIN_MSG_BODYLEN_OFFSET, realLen);
                // fill compressed data
                System.arraycopy(tmpData, 0,
                        dataBuf.array(), BIN_MSG_BODY_OFFSET, realLen);
                // fill attributes and MAGIC
                System.arraycopy(attr, 0, dataBuf.array(),
                        BIN_MSG_BODY_OFFSET + realLen, attr.length);

                out.write(dataBuf.array(), 0, dataBuf.capacity());
            } else {
                out.write(binMsgBuffer.array(), 0, binMsgBuffer.capacity());
            }

            writeMagic(out);
            out.close();
            return ByteBuffer.wrap(out.getData(), 0, out.getLength());
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }
add msg
@param attr
@param data
@return
inlong-common/src/main/java/org/apache/inlong/common/msg/InLongMsg.java