Converting Java to Avro and back when using kafka

Question

I'm using the confluent platform, 0.9.0.1 and kafka-avro-serializer 2.0.1. Trying to send events to kafka and read them back, I don't see how to turn events back into Java objects. I've read the avro and confluent docs, and there's hints that this is doable, but I can't see to find a good example. Here's my code, I get a GenericData$Record back when I read it with the KafkaConsumer, my question is how to get that back into a Java pojo. I found this bit of code that I used to serialize the object.

Here's my code:

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.ReflectDatumWriter;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.joda.time.DateTime;
import org.junit.Test;

import java.io.ByteArrayOutputStream;
import java.util.Collections;
import java.util.Properties;

/**
 * This is a test...
 */
public class KafkaAvroProducerTest {
    private static final Logger log = LogManager.getLogger(KafkaAvroProducerTest.class);

    @Test
    public void produceAndSendAndEvent() throws Exception {
        Properties props = new Properties();
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                org.apache.kafka.common.serialization.StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                io.confluent.kafka.serializers.KafkaAvroSerializer.class);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put("schema.registry.url", "http://localhost:8081");
        KafkaProducer producer = new KafkaProducer(props);

        log.debug("starting producer");
        String topic = "topic11";
        Schema schema = ReflectData.get().getSchema(Purchase.class);
        Purchase purchase = new Purchase("appStore", 9.99d, DateTime.now().getMillis(), "BRXh2lf9wm");

        ReflectDatumWriter<Purchase> reflectDatumWriter = new ReflectDatumWriter<>(schema);
        GenericDatumReader<Object> genericRecordReader = new GenericDatumReader<>(schema);
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        reflectDatumWriter.write(purchase, EncoderFactory.get().directBinaryEncoder(bytes, null));
        GenericRecord avroRecord = (GenericRecord) genericRecordReader.read(null, DecoderFactory.get().binaryDecoder(bytes.toByteArray(), null));
        ProducerRecord record = new ProducerRecord<Object, Object>(topic, avroRecord);

        Thread producerThread = new Thread(() -> {
            try {
                while(true) {
                    log.debug("send a message {}", record);
                    producer.send(record);
                    Thread.sleep(2000);
                }
            }catch(Exception ex) {
                log.error("error", ex);
            }
        });
        producerThread.start();

        props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "testGroup");
        props.put("auto.commit.enable", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        props.put("schema.registry.url", "http://localhost:8081");
        org.apache.kafka.clients.consumer.KafkaConsumer<String, GenericRecord> kafkaConsumer = new KafkaConsumer(props);
        kafkaConsumer.subscribe(Collections.singletonList(topic));

        Thread consumerThread = new Thread(() -> {
            try {
                while(true) {
                    try {
                        ConsumerRecords<String, GenericRecord> records = kafkaConsumer.poll(1000);
                        for (ConsumerRecord<String, GenericRecord> record1 : records) {//
                            log.debug("read - {}", record1.value().getClass());
                        }
                    }catch(Exception ex) {
                        log.error("error", ex);
                    }
                }
            }catch(Exception ex) {
                log.error("error", ex);
            }
        });
        consumerThread.start();
        System.in.read();
    }
}

Answer 1

As likely as the manually pulling your data out of a GenericRecord serializer into a java class... why write it manually when you can have a reflection lib do it for you?

For automatic conversion to a registered java type you'll be looking at creating your own KafkaAvroDeserializer that creates a SpecificRecord created through a ReflectDatumReader as listed in this stackoverflow post... - KafkaAvroDeserializer does not return SpecificRecord but returns GenericRecord

Answer 2

I never use Avro, but looking at https://avro.apache.org/docs/1.7.6/api/java/org/apache/avro/generic/GenericRecord.html why can't you simple populate your POJO manually...

class MyPojo {
    public int v1;
    public String v2;
}

// copied from your example code
ConsumerRecords<String, GenericRecord> records = kafkaConsumer.poll(1000);
for (ConsumerRecord<String, GenericRecord> record1 : records) {
    GenericRecord avroRecord = record1.value();
    MyPojo pojo = new MyPojo();
    pojo.v1 = (Integer)avroRecord.get("<fieldname1>");
    pojo.v2 = (String)avroRecord.get("<fieldname2>");

    // process current pojo
}

Not sure if this makes sense. If this works, I would move it into a constructor MyPojo(GenericRecord).