CODEWITHSUNDEEP
javaapache-kafka
Piyush Kumar
Piyush Kumar

Reputation: 113

Unable to push messages to apache kafka?

I am new to kafka and trying to run a sample apache java producer code to push data to kafka. I am able to create new topics through java but while pushing, I am getting an exception. Here is the code:

package kafkaTest;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Properties;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class HelloKafkaProducer {
final static String TOPIC = "test_kafka1";

public static void main(String[] argv){
    Properties properties = new Properties();
    properties.put("metadata.broker.list", "172.25.37.66:9092");
    ZkClient zkClient = new ZkClient("172.25.37.66:2181", 4000, 6000, new      BytesPushThroughSerializer());
    List<String> brokerList = zkClient.getChildren("/brokers/topics");
    for(int i=0;i<brokerList.size();i++){
        System.out.println(brokerList.get(i));
    }

   properties.put("zk.connect","172.25.37.66:2181");

    properties.put("serializer.class","kafka.serializer.StringEncoder");
    ProducerConfig producerConfig = new ProducerConfig(properties);

    kafka.javaapi.producer.Producer<String,String> producer = new kafka.javaapi.producer.Producer<String, String>(producerConfig);
    SimpleDateFormat sdf = new SimpleDateFormat();
    KeyedMessage<String, String> message =new KeyedMessage<String, String>(TOPIC,"Test message from java program " + sdf.format(new Date()));
    System.out.println(message);
    producer.send(message);
    /*Consumer consumerThread = new Consumer(TOPIC);
    consumerThread.start();*/
}
}

And this is the stacktrace :

topic1
test_kafka1
topic11
test
test_kafka
KeyedMessage(test_kafka1,null,null,Test message from java program 4/5/15 1:30 PM)
Exception in thread "main" [2015-05-04 13:30:41,432] ERROR Failed to send requests for topics test_kafka1 with correlation ids in [0,12] (kafka.producer.async.DefaultEventHandler:97)
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
at kafka.producer.Producer.send(Producer.scala:77)
at kafka.javaapi.producer.Producer.send(Producer.scala:33)
at kafkaTest.HelloKafkaProducer.main(HelloKafkaProducer.java:54)

On the console, I see the [2015-05-04 18:55:29,959] INFO Closing socket connection to /172.17.70.73. (kafka.network.Processor) everytime I run the program. I am able to push and pull using console to the topics.

All help would be appreciated . Thanks.

Upvotes: 0

Views: 1910

Answers (1)

Pranjal Sahu
Pranjal Sahu

Reputation: 1469

You don't connect to zookeeper in case of Kafka Producer. You have to connect to the Broker. For that purpose use the following property.

props.put("metadata.broker.list", "localhost:9092, broker1:9092");

Here I have used localhost in your case it will be 172.25.37.66

Upvotes: 2

Related Questions