Unable to produce messages to topic: failed to send producer

[Charlie Davis <ch...@mimosa.co>]

Hey everyone,

I’m hoping someone can help me with an issues I’m having. I’ll be using my console output so I’m sorry for the console spam. :)

So first, I list my topics:

?  kafka_2.11-0.8.2.0  bin/kafka-topics.sh --list --zookeeper localhost:6002
dog
?  kafka_2.11-0.8.2.0

And I have a topic called “dog.” Next, I do a describe..

?  kafka_2.11-0.8.2.0  bin/kafka-topics.sh --describe --zookeeper localhost:6002
Topic:dog PartitionCount:1 ReplicationFactor:1 Configs:
Topic: dog Partition: 0 Leader: 4 Replicas: 4 Isr: 4
?  kafka_2.11-0.8.2.0

Ok. Cool. Now, here’s where I run into problems, when I actually send a message:

?  kafka_2.11-0.8.2.0  bin/kafka-console-producer.sh --broker-list localhost:6001 --topic dog
[2015-02-09 17:14:45,060] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
This is a message so why don't you work?
[2015-02-09 17:14:50,235] WARN Failed to send producer request with correlation id 2 to broker 4 with data for partitions [dog,0] (kafka.producer.async.DefaultEventHandler)
java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:103)
at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:102)
at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at kafka.producer.SyncProducer.send(SyncProducer.scala:101)
at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:255)
at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:106)
at kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:100)
at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:778)
at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:99)
at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:230)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:99)
at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:777)
at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:100)
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:72)
at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88)
at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68)
at scala.collection.immutable.Stream.foreach(Stream.scala:594)
at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
[2015-02-09 17:14:50,354] WARN Failed to send producer request with correlation id 5 to broker 4 with data for partitions [dog,0] (kafka.producer.async.DefaultEventHandler)
java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:103)
at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
at kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:102)
at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
at kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)

And my kafka service has a pile of… info messages I guess?

nmskafka_1 | [2015-02-10 01:04:46,704] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:46,860] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:46,970] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:47,084] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:47,200] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:55,354] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:55,470] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:55,584] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:04:55,699] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:06:56,163] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:06:56,328] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)
nmskafka_1 | [2015-02-10 01:06:56,449] INFO Closing socket connection to /10.0.2.2. (kafka.network.Processor)

I’m not sure what’s wrong and haven’t a clue where to start digging. Anyone offer some advice?

Charlie Davis • Mimosa • Sr. Cloud Developer

This email may contain confidential and privileged material for the sole use of the intended recipient(s). Any review, use, distribution or disclosure by others is strictly prohibited. If you are not the intended recipient (or authorized to receive for the recipient), please contact the sender by reply email and delete all copies of this message.