@Test
    public void testParsingMalformedMessage() {
        MetadataRecordSerde serde = new MetadataRecordSerde();
        ByteBuffer buffer = ByteBuffer.allocate(4);
        buffer.put((byte) 0x01); // frame version
        buffer.put((byte) 0x00); // apiKey
        buffer.put((byte) 0x00); // apiVersion
        buffer.put((byte) 0x80); // malformed data
        buffer.position(0);
        buffer.limit(4);
        assertStartsWith("Failed to deserialize record with type",
                assertThrows(MetadataParseException.class,
                        () -> serde.read(new ByteBufferAccessor(buffer), buffer.remaining())).getMessage());
    }
Test attempting to parse an event which has a malformed message body.
metadata/src/test/java/org/apache/kafka/metadata/MetadataRecordSerdeTest.java