How to generate pojo from kafka avro record?

Question

I have a User class and I am seralizing it as avro, (using Confluent avro serializer and schema registry) and publish it to a Kafka topic. I made a consumer to print data to console and it works fine. What I am trying now is to create the original object from this data. For example, I am publishing "User" object as avro to Kafka topic. I am trying to recreate that user object (instead of console output) after consuming it. Is this possible?

Below is my code

User class

public class User { 
    int id;
    String name;    
    public User(){} 
    public User(int id, String name) {
        super();
        this.id = id;
        this.name = name;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }   
}

Consumer code

User user = new User();



Properties props = new Properties();

props.put("bootstrap.servers", "127.0.0.1:9092");

props.put("group.id", "avro-consumer-group");

props.put("key.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);

props.put("value.deserializer", io.confluent.kafka.serializers.KafkaAvroDeserializer.class);

props.put("schema.registry.url","http://127.0.0.1:8081");

KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<String, GenericRecord>(props);



consumer.subscribe(Arrays.asList("avrotesttopic"));

System.out.println("Subscribed to topic " + "avrotesttopic");



while (true) {

    ConsumerRecords<String, GenericRecord> records = consumer.poll(100);

    for (org.apache.kafka.clients.consumer.ConsumerRecord<String, GenericRecord> record : records){

        System.out.printf("value = %sn",record.value());    

        //output->  value = {"id": 10, "name": "testName"}

    }

}

Thanks

Answer 1

Considering that you are using KafkaAvroDeserializer, you will need to set the folloing property as part of your consumer configuration

    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");

it will allow you to get the SpecificRecord instead of GenericRecord already handled by the Consumer. Here is an example

https://dzone.com/articles/kafka-avro-serialization-and-the-schema-registry