@Override
    public CoordinatorRecord deserialize(
        ByteBuffer keyBuffer,
        ByteBuffer valueBuffer
    ) throws RuntimeException {
        final short recordType = readVersion(keyBuffer, "key");
        final ApiMessage keyMessage = apiMessageKeyFor(recordType);
        readMessage(keyMessage, keyBuffer, recordType, "key");

        if (valueBuffer == null) {
            return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
        }

        final ApiMessage valueMessage = apiMessageValueFor(recordType);
        final short valueVersion = readVersion(valueBuffer, "value");
        readMessage(valueMessage, valueBuffer, valueVersion, "value");

        return new CoordinatorRecord(
            new ApiMessageAndVersion(keyMessage, recordType),
            new ApiMessageAndVersion(valueMessage, valueVersion)
        );
    }
@Test
    public void testDeserializeWithInvalidRecordType() {
        GroupCoordinatorRecordSerde serde = new GroupCoordinatorRecordSerde();

        ByteBuffer keyBuffer = ByteBuffer.allocate(64);
        keyBuffer.putShort((short) 255);
        keyBuffer.rewind();

        ByteBuffer valueBuffer = ByteBuffer.allocate(64);

        CoordinatorLoader.UnknownRecordTypeException ex =
            assertThrows(CoordinatorLoader.UnknownRecordTypeException.class,
                () -> serde.deserialize(keyBuffer, valueBuffer));
        assertEquals((short) 255, ex.unknownType());
    }