@Override
    public Struct read(ByteBuffer buffer) {
        if (cachedStruct != null) {
            return cachedStruct;
        }
        Object[] objects = new Object[fields.length];
        for (int i = 0; i < fields.length; i++) {
            try {
                if (tolerateMissingFieldsWithDefaults) {
                    if (buffer.hasRemaining()) {
                        objects[i] = fields[i].def.type.read(buffer);
                    } else if (fields[i].def.hasDefaultValue) {
                        objects[i] = fields[i].def.defaultValue;
                    } else {
                        throw new SchemaException("Missing value for field '" + fields[i].def.name +
                                "' which has no default value.");
                    }
                } else {
                    objects[i] = fields[i].def.type.read(buffer);
                }
            } catch (Exception e) {
                throw new SchemaException("Error reading field '" + fields[i].def.name + "': " +
                                          (e.getMessage() == null ? e.getClass().getName() : e.getMessage()));
            }
        }
        return new Struct(this, objects);
    }
Read a struct from the buffer. If this schema is configured to tolerate missing
optional fields at the end of the buffer, these fields are replaced with their default
values; otherwise, if the schema does not tolerate missing fields, or if missing fields
don't have a default value, a {@code SchemaException} is thrown to signify that mandatory
fields are missing.
clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java