def decode_message(bytes)
      raise "Empty message" if bytes.empty?
      reserved, schema_id, avro = bytes.unpack('cNa*')
      raise "Reserved byte #{reserved.inspect} in message header (expected 0)!\nmessage: #{bytes.inspect}" unless 0 == reserved

      reader = reader_for(schema_id)
      decoder = Avro::IO::BinaryDecoder.new(StringIO.new(avro))

      wrap_error "parsing message with schema #{schema_id}" do
        [reader.writers_schema, reader.read(decoder)]
      end
    end
def decode_message(bytes)
      raise "Empty message" if bytes.empty?
      reserved, schema_id, avro = bytes.unpack('cNa*')
      raise "Reserved byte #{reserved.inspect} in message header (expected 0)!\nmessage: #{bytes.inspect}" unless 0 == reserved

      reader = reader_for(schema_id)
      decoder = Avro::IO::BinaryDecoder.new(StringIO.new(avro))

      wrap_error "parsing message with schema #{schema_id}" do
        [reader.writers_schema, reader.read(decoder)]
      end
    end
message format:

 Every message sent to Kafka is Avro-encoded, and prefixed with five
 bytes:
  - The first byte is always 0, and reserved for future use.
  - The next four bytes are the schema ID in big-endian byte order.

 per https://github.com/confluentinc/bottledwater-pg/blob/8a11825/kafka/registry.c#L6-L8