class RetainedMessageValueType extends BasicDataType<RetainedMessage> {
        // Layout for RetainedMessage:
        // - topic String
        // - qos int
        // - payload byte[]
        // - flag map to say if contains properties, expiry time (MSB, LSB)
        // - (opt) expiry time in epoch millis long
        // - (opt) properties

        private final PropertiesDataType propertiesDataType = new PropertiesDataType();

        private static final byte MESSAGE_EXPIRY_FLAG = 0x01;
        private static final byte PROPERTIES_FLAG = MESSAGE_EXPIRY_FLAG << 1;

        @Override
        public int getMemory(RetainedMessage retainedMsg) {
            int bytesSize = StringDataType.INSTANCE.getMemory(retainedMsg.getTopic().toString()) +
                1 + // qos, 1 byte
                4 + retainedMsg.getPayload().length + // length + bytes
                1; // flags
            if (retainedMsg.getExpiryTime() != null) {
                bytesSize += 8; // long
            }
            int propertiesSize = retainedMsg.getMqttProperties().length > 0 ?
                propertiesDataType.getMemory(retainedMsg.getMqttProperties()) :
                0;

            return bytesSize + propertiesSize;
        }

        @Override
        public void write(WriteBuffer buff, RetainedMessage retainedMsg) {
            StringDataType.INSTANCE.write(buff, retainedMsg.getTopic().toString());
            buff.put((byte) retainedMsg.qosLevel().value());
            buff.putInt(retainedMsg.getPayload().length);
            buff.put(retainedMsg.getPayload());

            byte flagsBitmask = 0x00;
            if (retainedMsg.getExpiryTime() != null) {
                flagsBitmask = (byte) (flagsBitmask | MESSAGE_EXPIRY_FLAG);
            }
            if (retainedMsg.getMqttProperties().length > 0) {
                flagsBitmask = (byte) (flagsBitmask | PROPERTIES_FLAG);
            }

            buff.put(flagsBitmask);

            if (retainedMsg.getExpiryTime() != null) {
                buff.putLong(retainedMsg.getExpiryTime().toEpochMilli());
            }

            if (retainedMsg.getMqttProperties().length > 0) {
                propertiesDataType.write(buff, retainedMsg.getMqttProperties());
            }
        }

        @Override
        public RetainedMessage read(ByteBuffer buff) {<FILL_FUNCTION_BODY>}

        @Override
        public RetainedMessage[] createStorage(int size) {
            return new RetainedMessage[size];
        }
    }

            final String topicStr = StringDataType.INSTANCE.read(buff);
            final MqttQoS qos = MqttQoS.valueOf(buff.get());

            final int payloadSize = buff.getInt();
            byte[] payload = new byte[payloadSize];
            buff.get(payload);

            final byte flags = buff.get();

            final Instant expiry;
            if ((flags & MESSAGE_EXPIRY_FLAG) > 0) {
                long millis = buff.getLong();
                expiry = Instant.ofEpochMilli(millis);
            } else {
                expiry = null;
            }

            final MqttProperties.MqttProperty[] mqttProperties;
            if ((flags & PROPERTIES_FLAG) > 0) {
                mqttProperties = propertiesDataType.read(buff);
            } else {
                mqttProperties = new MqttProperties.MqttProperty[0];
            }

            if ((flags & MESSAGE_EXPIRY_FLAG) > 0) {
                return new RetainedMessage(new Topic(topicStr), qos, payload, mqttProperties, expiry);
            } else {
                return new RetainedMessage(new Topic(topicStr), qos, payload, mqttProperties);
            }
        
<no_super_class>