@Test(expected = RuntimeException.class) public void incorrectMagicByte() { underTest.deserialize(TOPIC, new byte[] { 0x1 }); }
@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); } }
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); } } }
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); } }  }
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(); }
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; }