public static List<SortEvent> decodeCacheMessageBody(String inlongGroupId, String inlongStreamId,
            INLONG_COMPRESSED_TYPE compressedType, byte[] msgBody) throws IOException {
        // uncompress
        byte[] srcBytes = null;
        switch (compressedType) {
            case INLONG_SNAPPY:
                srcBytes = Snappy.uncompress(msgBody);
                break;
            case INLONG_GZ:
                srcBytes = GzipUtils.decompress(msgBody);
                break;
            case INLONG_NO_COMPRESS:
            default:
                srcBytes = msgBody;
                break;
        }
        // decode
        MessageObjs msgObjs = MessageObjs.parseFrom(srcBytes);
        List<SortEvent> events = new ArrayList<>(msgObjs.getMsgsList().size());
        for (MessageObj msgObj : msgObjs.getMsgsList()) {
            SortEvent event = new SortEvent(inlongGroupId, inlongStreamId, msgObj);
            events.add(event);
        }
        return events;
    }
decodeCacheMessageBody

@param  inlongGroupId
@param  inlongStreamId
@param  compressedType
@param  msgBody
@return List,SortEvent
@throws IOException
inlong-sdk/sdk-common/src/main/java/org/apache/inlong/sdk/commons/protocol/EventUtils.java