Odtnhj

I have customer-avro topic in kafka cluster with broker list kafka1.com:9092,kafka2.com:9092,kafka3.com:9092.

I'm able to produce record to the topic from java and retrieve it from console using below command.

 $ sudo bin/kafka-avro-console-consumer --bootstrap-server kafka1.com:9092,kafka2.com:9092,kafka3.com:9092 --topic customer-avro --property schema.registry.url=http://schemaregistry1.com:8081 --from-beginning
 OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
 "first_name":"Mottu","last_name":"kalidasan","age":24,"height":5.4,"weight":6.4,"automated_email":false

In the above command it is obvious there is a record in the topic. So I written below consumer logic to retrieve records.

 package com.example;

 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 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.common.serialization.StringDeserializer;
 import java.util.Collections;
 import java.util.Properties;

 public class KafkaConsumerV1 
 public static void main(String args) 
 Properties props = new Properties();
 props.setProperty("bootstrap.servers", "kafka1.com:9092,kafka2.com:9092,kafka3.com:9092");
 props.setProperty("group.id", "my-avro-consumer");
 props.setProperty("enable.auto.commit", "false");
 props.setProperty("auto.offset.reset", "earliest");
 props.setProperty("key.deserializer", StringDeserializer.class.getName());
 props.setProperty("value.deserializer", KafkaAvroDeserializer.class.getName());
 props.setProperty("schema.registry.url", "http://schemaregistry1.com:8081");
 props.setProperty("specific.avro.reader", "true");

 try(KafkaConsumer<String, Customer> consumer = new KafkaConsumer<String, Customer>(props)) 
 String topic = "customer-avro";
 consumer.subscribe(Collections.singleton(topic));
 System.out.println("Waiting for data...");
 while (true) 
 ConsumerRecords<String, Customer> records = consumer.poll(500);
 for (ConsumerRecord<String, Customer> record : records) 
 record.key();
 Customer customer = record.value();
 System.out.println("customer "+ customer);
 
 consumer.commitSync();

 
 
 
 

But the consumer is not retrieving records and it is always displaying "waiting for data".

What is the mistake in my java code. How can I retrieve data using java consumer.