You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Sa Li <sa...@gmail.com> on 2014/12/18 23:21:32 UTC
can't produce message in kafka production
Dear all
We just build a kafka production cluster, I can create topics in kafka
production from another host. But when I am send very simple message as
producer, it generate such errors:
root@precise64:/etc/kafka# bin/kafka-console-producer.sh --broker-list
10.100.98.100:9092 --topic my-replicated-topic-production
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in
[jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in
[jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
my test message 1
[2014-12-18 21:44:25,830] WARN Failed to send producer request with
correlation id 2 to broker 101 with data for partitions
[my-replicated-topic-production,1]
(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:256)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
at
scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at
scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
at
scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
at
kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
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:547)
at
kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
at
kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
[2014-12-18 21:44:25,948] WARN Failed to send producer request with
correlation id 5 to broker 101 with data for partitions
[my-replicated-topic-production,1]
(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:256)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
at
scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at
scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
at
scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
at
kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
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:547)
at
kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
at
kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
[2014-12-18 21:44:26,129] WARN Failed to send producer request with
correlation id 8 to broker 100 with data for partitions
[my-replicated-topic-production,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:256)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
at
kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
at
scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at
scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
at
scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
at
kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
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:547)
at
kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
at
kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
Any ideas about this type of error?
thanks
--
Alec Li
Re: can't produce message in kafka production
Posted by Neha Narkhede <ne...@confluent.io>.
The producer is complaining that it's socket channel is already closed.
Which makes me think it was closed due to some error that is not present in
your logs. I'd enable DEBUG and see if that shows the cause.
On Thu, Dec 18, 2014 at 4:13 PM, Gwen Shapira <gs...@cloudera.com> wrote:
>
> Perhaps you have the logs from broker? It may show other errors that
> can help us troubleshoot.
>
> On Thu, Dec 18, 2014 at 4:11 PM, Sa Li <sa...@gmail.com> wrote:
> > Thanks, Gwen, I telnet it,
> > root@precise64:/etc/kafka# telnet 10.100.98.100 9092
> > Trying 10.100.98.100...
> > Connected to 10.100.98.100.
> > Escape character is '^]'.
> >
> > seems it connected, and I check with system operation people, netstate
> > should 9092 is listening. I am assuming this is the connection issue,
> since
> > I can run the same command to my dev-cluster with no problem at all,
> which
> > is 10.100.70.128:9092.
> >
> > Just in case, is it possibly caused by other types of issues?
> >
> > thanks
> >
> > Alec
> >
> > On Thu, Dec 18, 2014 at 2:33 PM, Gwen Shapira <gs...@cloudera.com>
> wrote:
> >>
> >> Looks like you can't connect to: 10.100.98.100:9092
> >>
> >> I'd validate that this is the issue using telnet and then check the
> >> firewall / ipfilters settings.
> >>
> >> On Thu, Dec 18, 2014 at 2:21 PM, Sa Li <sa...@gmail.com> wrote:
> >> > Dear all
> >> >
> >> > We just build a kafka production cluster, I can create topics in kafka
> >> > production from another host. But when I am send very simple message
> as
> >> > producer, it generate such errors:
> >> >
> >> > root@precise64:/etc/kafka# bin/kafka-console-producer.sh
> --broker-list
> >> > 10.100.98.100:9092 --topic my-replicated-topic-production
> >> > SLF4J: Class path contains multiple SLF4J bindings.
> >> > SLF4J: Found binding in
> >> >
> >>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >> > SLF4J: Found binding in
> >> >
> >>
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> >> > SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> >> > explanation.
> >> > SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> >> > my test message 1
> >> > [2014-12-18 21:44:25,830] WARN Failed to send producer request with
> >> > correlation id 2 to broker 101 with data for partitions
> >> > [my-replicated-topic-production,1]
> >> > (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:256)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> >> > at
> >> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> >> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> >> > 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:547)
> >> > at
> >> >
> >>
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> >> > at
> >> >
> kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> >> > [2014-12-18 21:44:25,948] WARN Failed to send producer request with
> >> > correlation id 5 to broker 101 with data for partitions
> >> > [my-replicated-topic-production,1]
> >> > (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:256)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> >> > at
> >> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> >> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> >> > 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:547)
> >> > at
> >> >
> >>
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> >> > at
> >> >
> kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> >> > [2014-12-18 21:44:26,129] WARN Failed to send producer request with
> >> > correlation id 8 to broker 100 with data for partitions
> >> > [my-replicated-topic-production,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:256)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> >> > at
> >> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> >> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> >> > at
> >> >
> >>
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> >> > at
> >> >
> >>
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> >> > 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:547)
> >> > at
> >> >
> >>
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> >> > at
> >> >
> kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> >> >
> >> > Any ideas about this type of error?
> >> >
> >> >
> >> > thanks
> >> >
> >> > --
> >> >
> >> > Alec Li
> >>
> >
> >
> > --
> >
> > Alec Li
>
--
Thanks,
Neha
Re: can't produce message in kafka production
Posted by Gwen Shapira <gs...@cloudera.com>.
Perhaps you have the logs from broker? It may show other errors that
can help us troubleshoot.
On Thu, Dec 18, 2014 at 4:11 PM, Sa Li <sa...@gmail.com> wrote:
> Thanks, Gwen, I telnet it,
> root@precise64:/etc/kafka# telnet 10.100.98.100 9092
> Trying 10.100.98.100...
> Connected to 10.100.98.100.
> Escape character is '^]'.
>
> seems it connected, and I check with system operation people, netstate
> should 9092 is listening. I am assuming this is the connection issue, since
> I can run the same command to my dev-cluster with no problem at all, which
> is 10.100.70.128:9092.
>
> Just in case, is it possibly caused by other types of issues?
>
> thanks
>
> Alec
>
> On Thu, Dec 18, 2014 at 2:33 PM, Gwen Shapira <gs...@cloudera.com> wrote:
>>
>> Looks like you can't connect to: 10.100.98.100:9092
>>
>> I'd validate that this is the issue using telnet and then check the
>> firewall / ipfilters settings.
>>
>> On Thu, Dec 18, 2014 at 2:21 PM, Sa Li <sa...@gmail.com> wrote:
>> > Dear all
>> >
>> > We just build a kafka production cluster, I can create topics in kafka
>> > production from another host. But when I am send very simple message as
>> > producer, it generate such errors:
>> >
>> > root@precise64:/etc/kafka# bin/kafka-console-producer.sh --broker-list
>> > 10.100.98.100:9092 --topic my-replicated-topic-production
>> > SLF4J: Class path contains multiple SLF4J bindings.
>> > SLF4J: Found binding in
>> >
>> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>> > SLF4J: Found binding in
>> >
>> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
>> > SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
>> > explanation.
>> > SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
>> > my test message 1
>> > [2014-12-18 21:44:25,830] WARN Failed to send producer request with
>> > correlation id 2 to broker 101 with data for partitions
>> > [my-replicated-topic-production,1]
>> > (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:256)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
>> > at
>> >
>> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
>> > at
>> >
>> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
>> > at
>> >
>> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
>> > at
>> >
>> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
>> > at
>> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
>> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
>> > at
>> >
>> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
>> > 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:547)
>> > at
>> >
>> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
>> > at
>> > kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
>> > [2014-12-18 21:44:25,948] WARN Failed to send producer request with
>> > correlation id 5 to broker 101 with data for partitions
>> > [my-replicated-topic-production,1]
>> > (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:256)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
>> > at
>> >
>> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
>> > at
>> >
>> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
>> > at
>> >
>> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
>> > at
>> >
>> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
>> > at
>> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
>> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
>> > at
>> >
>> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
>> > 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:547)
>> > at
>> >
>> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
>> > at
>> > kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
>> > [2014-12-18 21:44:26,129] WARN Failed to send producer request with
>> > correlation id 8 to broker 100 with data for partitions
>> > [my-replicated-topic-production,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:256)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
>> > at
>> >
>> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
>> > at
>> >
>> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
>> > at
>> >
>> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
>> > at
>> >
>> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
>> > at
>> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
>> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
>> > at
>> >
>> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
>> > at
>> >
>> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
>> > 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:547)
>> > at
>> >
>> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
>> > at
>> > kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
>> >
>> > Any ideas about this type of error?
>> >
>> >
>> > thanks
>> >
>> > --
>> >
>> > Alec Li
>>
>
>
> --
>
> Alec Li
Re: can't produce message in kafka production
Posted by Sa Li <sa...@gmail.com>.
Thanks, Gwen, I telnet it,
root@precise64:/etc/kafka# telnet 10.100.98.100 9092
Trying 10.100.98.100...
Connected to 10.100.98.100.
Escape character is '^]'.
seems it connected, and I check with system operation people, netstate
should 9092 is listening. I am assuming this is the connection issue, since
I can run the same command to my dev-cluster with no problem at all, which
is 10.100.70.128:9092.
Just in case, is it possibly caused by other types of issues?
thanks
Alec
On Thu, Dec 18, 2014 at 2:33 PM, Gwen Shapira <gs...@cloudera.com> wrote:
>
> Looks like you can't connect to: 10.100.98.100:9092
>
> I'd validate that this is the issue using telnet and then check the
> firewall / ipfilters settings.
>
> On Thu, Dec 18, 2014 at 2:21 PM, Sa Li <sa...@gmail.com> wrote:
> > Dear all
> >
> > We just build a kafka production cluster, I can create topics in kafka
> > production from another host. But when I am send very simple message as
> > producer, it generate such errors:
> >
> > root@precise64:/etc/kafka# bin/kafka-console-producer.sh --broker-list
> > 10.100.98.100:9092 --topic my-replicated-topic-production
> > SLF4J: Class path contains multiple SLF4J bindings.
> > SLF4J: Found binding in
> >
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> > SLF4J: Found binding in
> >
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> > SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> > explanation.
> > SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> > my test message 1
> > [2014-12-18 21:44:25,830] WARN Failed to send producer request with
> > correlation id 2 to broker 101 with data for partitions
> > [my-replicated-topic-production,1]
> > (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:256)
> > at
> >
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> > at
> >
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> > at
> >
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> > at
> >
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> > at
> >
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> > at
> >
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> > at
> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> > at
> >
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> > at
> >
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> > 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:547)
> > at
> >
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> > at
> > kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> > [2014-12-18 21:44:25,948] WARN Failed to send producer request with
> > correlation id 5 to broker 101 with data for partitions
> > [my-replicated-topic-production,1]
> > (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:256)
> > at
> >
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> > at
> >
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> > at
> >
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> > at
> >
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> > at
> >
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> > at
> >
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> > at
> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> > at
> >
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> > at
> >
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> > 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:547)
> > at
> >
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> > at
> > kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> > [2014-12-18 21:44:26,129] WARN Failed to send producer request with
> > correlation id 8 to broker 100 with data for partitions
> > [my-replicated-topic-production,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:256)
> > at
> >
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> > at
> >
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> > at
> >
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> > at
> >
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> > at
> >
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> > at
> >
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> > at
> scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> > at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> > at
> >
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> > at
> >
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> > 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:547)
> > at
> >
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> > at
> > kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> >
> > Any ideas about this type of error?
> >
> >
> > thanks
> >
> > --
> >
> > Alec Li
>
--
Alec Li
Re: can't produce message in kafka production
Posted by Gwen Shapira <gs...@cloudera.com>.
Looks like you can't connect to: 10.100.98.100:9092
I'd validate that this is the issue using telnet and then check the
firewall / ipfilters settings.
On Thu, Dec 18, 2014 at 2:21 PM, Sa Li <sa...@gmail.com> wrote:
> Dear all
>
> We just build a kafka production cluster, I can create topics in kafka
> production from another host. But when I am send very simple message as
> producer, it generate such errors:
>
> root@precise64:/etc/kafka# bin/kafka-console-producer.sh --broker-list
> 10.100.98.100:9092 --topic my-replicated-topic-production
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.6.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in
> [jar:file:/etc/kafka/core/build/dependant-libs-2.10.4/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an
> explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> my test message 1
> [2014-12-18 21:44:25,830] WARN Failed to send producer request with
> correlation id 2 to broker 101 with data for partitions
> [my-replicated-topic-production,1]
> (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:256)
> at
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> at
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> at
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> at
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> at
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> at
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> 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:547)
> at
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> at
> kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> [2014-12-18 21:44:25,948] WARN Failed to send producer request with
> correlation id 5 to broker 101 with data for partitions
> [my-replicated-topic-production,1]
> (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:256)
> at
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> at
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> at
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> at
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> at
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> at
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> 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:547)
> at
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> at
> kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
> [2014-12-18 21:44:26,129] WARN Failed to send producer request with
> correlation id 8 to broker 100 with data for partitions
> [my-replicated-topic-production,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:256)
> at
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:107)
> at
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:99)
> at
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> at
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
> at
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:226)
> at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
> at scala.collection.mutable.HashMap.foreach(HashMap.scala:98)
> at
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771)
> at
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:99)
> 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:547)
> at
> kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
> at
> kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
>
> Any ideas about this type of error?
>
>
> thanks
>
> --
>
> Alec Li