DataDeserializer implements Deserializer<Record> { @Override public Record deserialize(String topic, byte[] data) { ByteBuffer buffer = ByteBuffer.wrap(data); if (buffer.get() != MAGIC_BYTE) { throw new RuntimeException("Unknown magic byte!"); } int version = buffer.getInt(); Schema schema = schemaLookup.getSchema(version); int offset = buffer.position(); int length = buffer.remaining(); DatumReader<Record> reader = new GenericDatumReader<>(schema, schema, genericData); Decoder decoder = DecoderFactory.get().binaryDecoder(data, offset, length, null); try { return reader.read(null, decoder); } catch (IOException e) { throw new RuntimeException("Unable to decode record.", e); } }  @Override Record deserialize(String topic, byte[] data); @Override void configure(Map<String, ?> configs, boolean isKey); @Override void close(); static final byte MAGIC_BYTE; }
@Test public void validSchemaAndData() { Schema schema = SchemaBuilder.record("r").fields().name("int8").type().intType().noDefault().endRecord(); Record record = new Record(schema); record.put("int8", 2); int version = 1; when(schemaLookup.getSchema(1)).thenReturn(schema); byte[] value = toAvroBinary(schema, record, version); Record result = underTest.deserialize(TOPIC, value); assertThat(result.getSchema(), is(schema)); assertThat(result, is(record)); }
@Test(expected = RuntimeException.class) public void incorrectMagicByte() { underTest.deserialize(TOPIC, new byte[] { 0x1 }); }