SqlAvroSerde implements Serde<AvroData> { @Override public AvroData fromBytes(byte[] bytes) { GenericRecord data; try { data = reader.read(null, DecoderFactory.get().binaryDecoder(bytes, null)); return getAvroData(data, avroSchema); } catch (IOException e) { String errMsg = "Cannot decode message."; log.error(errMsg, e); throw new SamzaException(errMsg, e); } } SqlAvroSerde(Schema avroSchema); @Override AvroData fromBytes(byte[] bytes); @Override byte[] toBytes(AvroData object);  }
@Test public void testSqlAvroSerdeDeserialization() throws IOException { AvroData decodedDatum = (AvroData)serde.fromBytes(encodeMessage(sampleOrderRecord(), orderSchema)); Assert.assertTrue(decodedDatum.schema().getType() == org.apache.samza.sql.api.data.Schema.Type.STRUCT); Assert.assertTrue(decodedDatum.getFieldData("id").schema().getType() == org.apache.samza.sql.api.data.Schema.Type.INTEGER); Assert.assertTrue(decodedDatum.getFieldData("quantity").schema().getType() == org.apache.samza.sql.api.data.Schema.Type.INTEGER); Assert.assertTrue(decodedDatum.getFieldData("product").schema().getType() == org.apache.samza.sql.api.data.Schema.Type.STRING); }