AvroEventDeserializer implements EventDeserializer { @Override public Event readEvent() throws IOException { if (fileReader.hasNext()) { record = fileReader.next(record); out.reset(); datumWriter.write(record, encoder); encoder.flush(); Event event = EventBuilder.withBody(out.toByteArray()); if (schemaType == AvroSchemaType.HASH) { event.getHeaders().put(AVRO_SCHEMA_HEADER_HASH, schemaHashString); } else { event.getHeaders().put(AVRO_SCHEMA_HEADER_LITERAL, schema.toString()); } return event; } return null; } private  AvroEventDeserializer(Context context, ResettableInputStream ris); @Override Event readEvent(); @Override List<Event> readEvents(int numEvents); @Override void mark(); @Override void reset(); @Override void close(); static final String CONFIG_SCHEMA_TYPE_KEY; static final String AVRO_SCHEMA_HEADER_HASH; static final String AVRO_SCHEMA_HEADER_LITERAL; }
@Test public void testSchemaHash() throws IOException, NoSuchAlgorithmException { File tempFile = newTestFile(true); String target = tempFile.getAbsolutePath(); logger.info("Target: {}", target); TransientPositionTracker tracker = new TransientPositionTracker(target); Context context = new Context(); context.put(AvroEventDeserializer.CONFIG_SCHEMA_TYPE_KEY, AvroEventDeserializer.AvroSchemaType.HASH.toString()); ResettableInputStream in = new ResettableFileInputStream(tempFile, tracker); EventDeserializer des = new AvroEventDeserializer.Builder().build(context, in); Event event = des.readEvent(); String eventSchemaHash = event.getHeaders().get(AvroEventDeserializer.AVRO_SCHEMA_HEADER_HASH); String expectedSchemaHash = Hex.encodeHexString( SchemaNormalization.parsingFingerprint("CRC-64-AVRO", schema)); Assert.assertEquals(expectedSchemaHash, eventSchemaHash); }
@Test public void testSchemaLiteral() throws IOException { File tempFile = newTestFile(true); String target = tempFile.getAbsolutePath(); logger.info("Target: {}", target); TransientPositionTracker tracker = new TransientPositionTracker(target); Context context = new Context(); context.put(AvroEventDeserializer.CONFIG_SCHEMA_TYPE_KEY, AvroEventDeserializer.AvroSchemaType.LITERAL.toString()); ResettableInputStream in = new ResettableFileInputStream(tempFile, tracker); EventDeserializer des = new AvroEventDeserializer.Builder().build(context, in); Event event = des.readEvent(); String eventSchema = event.getHeaders().get(AvroEventDeserializer.AVRO_SCHEMA_HEADER_LITERAL); Assert.assertEquals(schema.toString(), eventSchema); }