CODEWITHSUNDEEP
javaapache-kafkakafka-consumer-apikafka-producer-api
z22
z22

Reputation: 10083

kafka produce and send message from java app and consume in cli

I am new to kafka. I am trying to send message through java app and consume it in command line prompt, but the message is not getting displayed on CLI.

Following is the java code:

    package com.kafka.prj;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;



public class KafkaProd {

    private static KafkaProducer<String, String> producer;

    public void initialize() {

        Properties props = new Properties();
         props.put("bootstrap.servers", "localhost: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");

//          ProducerConfig producerConfig = new ProducerConfig(producerProps);
          producer = new KafkaProducer<String, String>(props);

    }
    public void publishMesssage() throws Exception{            


        producer.send(new ProducerRecord<String, String>("test1", "dummy text msg"));


      return;
    }

    public static void main(String[] args) {

         KafkaProd kafkaProducer = new KafkaProd();
         // Initialize producer
         kafkaProducer.initialize();            
         // Publish message
         try {
            kafkaProducer.publishMesssage();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
         //Close the producer
         producer.close();

    }

}

In CLI, following is the command am using to consume the message sent in code above:

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test1 --from-beginning

The above command displays nothing, no error, no output.

Where am I getting wrong?

Upvotes: 1

Views: 2977

Answers (1)

Binita Bharati
Binita Bharati

Reputation: 5898

I was having the same problem, ie messages produced by kafka-console-producer.sh were visible on the kafka-console-consumer.sh console. But, on using the Java producer, kafka-console-consumer.sh console did not receive any messages. Also, the logs on the Java producer had this as the last line : 

2017-07-10 16:36:42 INFO  KafkaProducer:972 - Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.

This means that the Java producer is not able to connect to the Kafka brokers as given by the bootstrap.servers config on the Java producer. (Although, I was actually able to telnet to the broker port from the same machine as the Java producer ). Solution is to add the property : advertised.host.name on all the brokers. and make it equal to the IP/hostname of the broker. This should be in line with whatever you are providing in the bootstrap.servers.

My bootstrap.servers had the value - 192.168.10.12:9092,192.168.10.13:9092,192.168.10.14:9092, so on each of the brokers, I made advertised.host.name=192.168.10.12, advertised.host.name=192.168.10.13 and advertised.host.name=192.168.10.14 respectively.

Upvotes: 1

Related Questions