docs: @Override
    public GenericRecord parse(ByteBuffer bytes)
    {
      if (bytes.remaining() < 5) {
        throw new ParseException("record must have at least 5 bytes carrying version and schemaId");
      }
  
      byte version = bytes.get();
      if (version != V1) {
        throw new ParseException("found record of arbitrary version [%s]", version);
      }
  
      int schemaId = bytes.getInt();
      Schema schemaObj = schemaObjs.get(schemaId);
      if (schemaObj == null) {
        throw new ParseException("Failed to find schema for id [%s]", schemaId);
      }
  
      DatumReader<GenericRecord> reader = new GenericDatumReader<>(schemaObj);
      try (ByteBufferInputStream inputStream = new ByteBufferInputStream(Collections.singletonList(bytes))) {
        return reader.read(null, DecoderFactory.get().binaryDecoder(inputStream, null));
      }
      catch (EOFException eof) {
        // waiting for avro v1.9.0 (#AVRO-813)
        throw new ParseException(
            eof,
            "Avro's unnecessary EOFException, detail: [%s]",
            "https://issues.apache.org/jira/browse/AVRO-813"
        );
      }
      catch (Exception e) {
        throw new ParseException(e, "Fail to decode avro message with schemaId [%s].", schemaId);
      }
    }