public static MessagePack encodeSdkEvents(String inlongGroupId, String inlongStreamId,
            INLONG_COMPRESSED_TYPE compressedType, List<SdkEvent> events) throws IOException {
        // MessageObjs
        MessageObjs.Builder objsBuilder = MessageObjs.newBuilder();
        for (SdkEvent event : events) {
            MessageObj.Builder objBuilder = MessageObj.newBuilder();
            objBuilder.setMsgTime(event.getMsgTime());
            objBuilder.setSourceIp(event.getSourceIp());
            objBuilder.setBody(ByteString.copyFrom(event.getBody()));
            objsBuilder.addMsgs(objBuilder.build());
        }
        MessageObjs objs = objsBuilder.build();
        byte[] srcBytes = objs.toByteArray();
        byte[] compressedBytes = null;
        switch (compressedType) {
            case INLONG_SNAPPY:
                compressedBytes = Snappy.compress(srcBytes);
                break;
            case INLONG_GZ:
                compressedBytes = GzipUtils.compress(srcBytes);
                break;
            case INLONG_NO_COMPRESS:
            default:
                compressedBytes = srcBytes;
                break;
        }
        // MessagePack
        MessagePack.Builder packBuilder = MessagePack.newBuilder();
        packBuilder.setCompressBytes(ByteString.copyFrom(compressedBytes));
        // MessagePackHeader
        MessagePackHeader.Builder headerBuilder = MessagePackHeader.newBuilder();
        // string inlongGroupId = 1; //inlongGroupId
        headerBuilder.setInlongGroupId(inlongGroupId);
        // string inlongStreamId = 2; //inlongStreamId
        headerBuilder.setInlongStreamId(inlongStreamId);
        // int64 packId = 3; //pack id
        headerBuilder.setPackId(0);
        // int64 packTime = 4; //pack time, milliseconds
        headerBuilder.setPackTime(System.currentTimeMillis());
        // int32 msgCount = 5; //message count
        headerBuilder.setMsgCount(events.size());
        // int32 srcLength = 6; //total length of raw messages body
        headerBuilder.setSrcLength(srcBytes.length);
        // int32 compressLen = 7; //compress length of messages
        headerBuilder.setCompressLen(compressedBytes.length);
        // INLONG_COMPRESSED_TYPE compressType = 8; //compress type
        headerBuilder.setCompressType(compressedType);
        // map<string, string> params = 9; //additional parameters
        packBuilder.setHeader(headerBuilder.build());
        return packBuilder.build();
    }
encode

@param  inlongGroupId
@param  inlongStreamId
@param  compressedType
@param  events
@return MessagePack
@throws IOException
inlong-sdk/sdk-common/src/main/java/org/apache/inlong/sdk/commons/protocol/EventUtils.java