CODEWITHSUNDEEP
javaapache-kafka
user3623460
user3623460

Reputation: 103

No response from Kafka consumer

I am new to the Kafka world, I am trying to create simple java Kafka application,using an existing example of a Producer and Consumer.

The Producer gives the result :

Sent:Message 0
Sent:Message 1
Sent:Message 2
Sent:Message 3
.............

But on the Consumer side,the code execute correctly without errors but no results appeared on the console Consumer console

The Producer:

public class ProducerTest {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.33.10:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = null;
try {
  producer = new KafkaProducer<>(props);
  for (int i = 0; i < 100; i++) {
    String msg = "Message " + i;
    producer.send(new ProducerRecord<String, String>("HelloKafkaTopic", msg));
    System.out.println("Sent:" + msg);
  }
} catch (Exception e) {
  e.printStackTrace();

} finally {
  producer.close();
}

}

}

The Consumer:

 public class ConsumerTest {

public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.33.10:9092");
props.put("group.id", "group-1");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("auto.offset.reset", "smallest");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
kafkaConsumer.subscribe(Arrays.asList("HelloKafkaTopic"));
while (true) {
  ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    System.out.println("Partition: " + record.partition() + " Offset: " + record.offset()
        + " Value: " + record.value() + " ThreadID: " + Thread.currentThread().getId());
  }
}

 }  }

Any suggestions , Thank you in advance

Upvotes: 0

Views: 1054

Answers (1)

amethystic
amethystic

Reputation: 7089

Some suggestions:
1. Run kafka-console-consumer script with --from-beginning to verify if you could see the produced messages.
2. Change props.put("auto.offset.reset", "smallest"); to props.put("auto.offset.reset", "earliest");
3. Do not use such a small timeout when calling poll. Instead, use a relatively large timeout, such as 3000. Even using Integer.MAX_VALUE is reasonable.

Upvotes: 1

Related Questions