Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java (Java)

Code before:
        @Override
        public int sizeOf(Object o) {
            return 8;
        }

        @Override
        public String typeName() {
            return "FLOAT64";
        }

        @Override
        public Double validate(Object item) {
            if (item instanceof Double)
                return (Double) item;
            else
                throw new SchemaException(item + " is not a Double.");
        }

        @Override
        public String documentation() {
            return "Represents a double-precision 64-bit format IEEE 754 value. " +
                    "The values are encoded using eight bytes in network byte order (big-endian).";
        }
    };

    public String stringRead(ByteBuffer buffer, short length) {
        if (length > buffer.remaining())
            throw new SchemaException("Error reading string of length " + length + ", only " + buffer.remaining() + " bytes available");
        String result = Utils.utf8(buffer, length);
        buffer.position(buffer.position() + length);
        return result;
    }

    public static final DocumentedType STRING = new DocumentedType() {
        @Override
        public void write(ByteBuffer buffer, Object o) {
            byte[] bytes = Utils.utf8((String) o);
            if (bytes.length > Short.MAX_VALUE)
                throw new SchemaException("String length " + bytes.length + " is larger than the maximum string length.");
            buffer.putShort((short) bytes.length);
            buffer.put(bytes);
        }

        @Override
        public String read(ByteBuffer buffer) {
            short length = buffer.getShort();
            if (length < 0)
                throw new SchemaException("String length " + length + " cannot be negative");
            return stringRead(buffer, length);
        }


Review comment (suggestion):
Could this be static? Also, it seems that we can just reuse compactStringRead(). It's useful to do the length > Short.MAX_VALUE validation for both compact and non-compact string.

Code after:
                throw new SchemaException(item + " is not a Uuid.");
        }

        @Override
        public String documentation() {
            return "Represents a type 4 immutable universally unique identifier (Uuid). " +
                    "The values are encoded using sixteen bytes in network byte order (big-endian).";
        }
    };

    public static final DocumentedType FLOAT64 = new DocumentedType() {
        @Override
        public void write(ByteBuffer buffer, Object o) {
            ByteUtils.writeDouble((Double) o, buffer);
        }

        @Override
        public Object read(ByteBuffer buffer) {
            return ByteUtils.readDouble(buffer);
        }

        @Override
        public int sizeOf(Object o) {
            return 8;
        }

        @Override
        public String typeName() {
            return "FLOAT64";
        }

        @Override
        public Double validate(Object item) {
            if (item instanceof Double)
                return (Double) item;
            else
                throw new SchemaException(item + " is not a Double.");
        }

        @Override
        public String documentation() {
            return "Represents a double-precision 64-bit format IEEE 754 value. " +
                    "The values are encoded using eight bytes in network byte order (big-endian).";
        }
    };

    public static final DocumentedType STRING = new DocumentedType() {
        @Override
        public void write(ByteBuffer buffer, Object o) {
            byte[] bytes = Utils.utf8((String) o);
            if (bytes.length > Short.MAX_VALUE)