@Override
    protected void encode(ChannelHandlerContext ctx,
            EncodeObject encObject, List<Object> out) throws Exception {
        ByteBuf buf = null;
        int totalLength;
        try {
            if (encObject.getMsgType() == MsgType.MSG_ACK_SERVICE) {
                // msgType(1) + bodyLength(4) + attrsLength(4)
                totalLength = 1 + 4 + 4 + encObject.getMsgSize();
                buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
                buf.writeInt(totalLength);
                buf.writeByte(encObject.getIntMsgType());
                buf.writeInt(encObject.getBodyDataLength());
                if (encObject.getBodyDataLength() > 0) {
                    buf.writeBytes(encObject.getBodyData());
                }
                buf.writeInt(encObject.getAttrDataLength());
                if (encObject.getAttrDataLength() > 0) {
                    buf.writeBytes(encObject.getAttrData());
                }
            } else if (encObject.getMsgType() == MsgType.MSG_MULTI_BODY) {
                // msgType(1) + bodyLength(4) + attrsLength(4)
                totalLength = 1 + 4 + 4 + encObject.getMsgSize();
                buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
                buf.writeInt(totalLength);
                buf.writeByte(encObject.getIntMsgType());
                buf.writeInt(encObject.getBodyDataLength());
                if (encObject.getBodyDataLength() > 0) {
                    buf.writeBytes(encObject.getBodyData());
                }
                buf.writeInt(encObject.getAttrDataLength());
                if (encObject.getAttrDataLength() > 0) {
                    buf.writeBytes(encObject.getAttrData());
                }
            } else if (encObject.getMsgType() == MsgType.MSG_BIN_MULTI_BODY) {
                // msgType(1) + groupNum(2) + streamNum(2) + extField(2)
                // + dataTime(4) + msgCnt(2) + uniqueId(4) + bodyLength(4)
                // + attrsLength(2) + magic(2)
                totalLength = 1 + 2 + 2 + 2 + 4 + 2 + 4 + 4 + 2 + 2 + encObject.getMsgSize();
                buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
                buf.writeInt(totalLength);
                buf.writeByte(encObject.getIntMsgType());
                buf.writeShort(encObject.getGroupIdNum());
                buf.writeShort(encObject.getStreamIdNum());
                buf.writeShort(encObject.getExtField());
                buf.writeInt((int) encObject.getDtMs());
                buf.writeShort(encObject.getMsgCnt());
                buf.writeInt(encObject.getMessageId());
                buf.writeInt(encObject.getBodyDataLength());
                if (encObject.getBodyDataLength() > 0) {
                    buf.writeBytes(encObject.getBodyData());
                }
                buf.writeShort(encObject.getAttrDataLength());
                if (encObject.getAttrDataLength() > 0) {
                    buf.writeBytes(encObject.getAttrData());
                }
                buf.writeShort(0xee01);
            } else if (encObject.getMsgType() == MsgType.MSG_BIN_HEARTBEAT) {
                // msgType(1) + dataTime(4) + version(1) + bodyLength(4)
                // + attrsLength(2) + magic(2)
                totalLength = 1 + 4 + 1 + 4 + 2 + encObject.getAttrDataLength() + 2;
                buf = ByteBufAllocator.DEFAULT.buffer(4 + totalLength);
                buf.writeInt(totalLength);
                buf.writeByte(encObject.getIntMsgType());
                buf.writeInt((int) encObject.getDtMs());
                buf.writeByte(2);
                buf.writeInt(0);
                buf.writeShort(encObject.getAttrDataLength());
                if (encObject.getAttrDataLength() > 0) {
                    buf.writeBytes(encObject.getAttrData());
                }
                buf.writeShort(0xee01);
            }
        } catch (Throwable ex) {
            if (exptCounter.shouldPrint()) {
                logger.warn("ProtocolEncoder encode({}) message failure", encObject.getMsgType(), ex);
            }
        }
        if (buf != null) {
            out.add(buf);
        } else {
            if (exptCounter.shouldPrint()) {
                logger.warn("ProtocolEncoder write({}) buffer is null!", encObject.getMsgType());
            }
        }
    }
TCP protocol encoder class

Used to encode the request package sent to DataProxy
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/codec/ProtocolEncoder.java