logstash ship data to kafka occur error

[songyancui]

logstash error message as follow:

Output thread exception {:plugin=><LogStash::Outputs::Kafka host=>"127.0.0.1", topic=>"logstash">, :exception=>#<Kafka::SocketError: cannot write: Broken pipe>, :backtrace=>["/home/csy/dev/src/logstash/logstash/vendor/bundle/jruby/1.9/gems/kafka-rb-0.0.12/lib/kafka/io.rb:53:in rescue in write'", "/home/csy/dev/src/logstash/logstash/vendor/bundle/jruby/1.9/gems/kafka-rb-0.0.12/lib/kafka/io.rb:49:inwrite'", "/home/csy/dev/src/logstash/logstash/vendor/bundle/jruby/1.9/gems/kafka-rb-0.0.12/lib/kafka/producer.rb:32:in send'", "/home/csy/dev/src/logstash/logstash/lib/logstash/outputs/kafka.rb:36:inreceive'", "/home/csy/dev/src/logstash/logstash/lib/logstash/outputs/base.rb:55:in handle'", "/home/csy/dev/src/logstash/logstash/lib/logstash/agent.rb:765:inrun_output'", "/home/csy/dev/src/logstash/logstash/lib/logstash/agent.rb:386:in `block in start_output'"], :level=>:warn}

[2013-12-06 03:22:23,851] ERROR Closing socket for /127.0.0.1 because of error (kafka.network.Processor)
java.nio.BufferUnderflowException
at java.nio.HeapByteBuffer.get(HeapByteBuffer.java:127)
at java.nio.ByteBuffer.get(ByteBuffer.java:675)
at kafka.api.ApiUtils$.readShortString(ApiUtils.scala:38)
at kafka.api.ProducerRequest$.readFrom(ProducerRequest.scala:33)
at kafka.api.RequestKeys$$anonfun$1.apply(RequestKeys.scala:34)
at kafka.api.RequestKeys$$anonfun$1.apply(RequestKeys.scala:34)
at kafka.network.RequestChannel$Request.(RequestChannel.scala:49)
at kafka.network.Processor.read(SocketServer.scala:353)
at kafka.network.Processor.run(SocketServer.scala:245)
at java.lang.Thread.run(Thread.java:619)

my config is :

output{
kafka {
host => "127.0.0.1"
port => 9092
topic => "logstash"
}
}