Sumit,You need to mention arguments as well.You are passing empty list.for ex:
props.put("bootstrap.servers", "localhost:9092")
From: Sumit Khanna <[email protected]>
To: [email protected]
Sent: Tuesday, 12 July 2016 4:49 PM
Subject: kafka unable to send records - scala / spark
Hello Guys.
Have tried a lot, from kafka.javaapi. etc to Producer to KafkaProducer, and
am working with 0.9.0.0
This is the error I am getting :
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata
after 62 ms.
at
org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:437)
at
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:352)
at
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:248)
at
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:352)
at
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:248)
Also,
here is the kafka props :
var props = new Properties()
props.put("bootstrap.servers", "")
props.put("metadata.broker.list", "")
props.put("group.id", "")
props.put("producer.type", "")
props.put("key.serializer", "")
props.put("value.serializer", "")
props.put("request.required.acks", "1")
props.put("auto.create.topics.enable","true")
props.put("block.on.buffer.full","false")
val producer = new KafkaProducer[String,String](props)
partitionOfRecords.foreach
{
case x:String=>{
val message=new ProducerRecord[String,
String]("[TOPIC#1]
"+dbname+"_"+dbtable,dbname,x)
producer.send(message).get();
}
}
}
Please help.
Thanks in advance.
Best,
Sumit Khanna
