@Test public void testSerializeAvro() throws Exception { Schema schema = new Schema.Parser().parse("{\n" + "\t\"type\":\"record\",\n" + "\t\"name\":\"test_schema_1\",\n" + "\t\"fields\" : [ {\n" + "\t\t\"name\":\"a\",\n" + "\t\t\"type\":\"int\"\n" + "\t\t},\n" + "\t\t{\n" + "\t\t\"name\":\"b\",\n" + "\t\t\"type\":\"string\"\n" + "\t\t}\n" + "\t\t]\n" + "}"); GenericRecord event = new GenericData.Record(schema); event.put("a", 1); event.put("b", "static string"); byte[] m = BaseProducer.serializeAvro(schema,event); DatumReader<GenericData.Record> reader = new GenericDatumReader<GenericData.Record>(schema); BinaryDecoder binaryDecoder = DecoderFactory.get().binaryDecoder(m,null); GenericRecord record = reader.read(null,binaryDecoder); Assert.assertEquals(record,event); }
public static byte[] serializeAvro(Schema schema, GenericRecord event) throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(stream, null); DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema); datumWriter.write(event, binaryEncoder); binaryEncoder.flush(); IOUtils.closeQuietly(stream); return stream.toByteArray(); }
BaseProducer { public static byte[] serializeAvro(Schema schema, GenericRecord event) throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(stream, null); DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema); datumWriter.write(event, binaryEncoder); binaryEncoder.flush(); IOUtils.closeQuietly(stream); return stream.toByteArray(); } }
BaseProducer { public static byte[] serializeAvro(Schema schema, GenericRecord event) throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(stream, null); DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema); datumWriter.write(event, binaryEncoder); binaryEncoder.flush(); IOUtils.closeQuietly(stream); return stream.toByteArray(); } BaseProducer(Properties props); }
BaseProducer { public static byte[] serializeAvro(Schema schema, GenericRecord event) throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(stream, null); DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema); datumWriter.write(event, binaryEncoder); binaryEncoder.flush(); IOUtils.closeQuietly(stream); return stream.toByteArray(); } BaseProducer(Properties props); static byte[] serializeAvro(Schema schema, GenericRecord event); void publish(GenericRecord event, String topic, Schema schema); void close(); }
BaseProducer { public static byte[] serializeAvro(Schema schema, GenericRecord event) throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(stream, null); DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema); datumWriter.write(event, binaryEncoder); binaryEncoder.flush(); IOUtils.closeQuietly(stream); return stream.toByteArray(); } BaseProducer(Properties props); static byte[] serializeAvro(Schema schema, GenericRecord event); void publish(GenericRecord event, String topic, Schema schema); void close();  }