kafka broker error

[2015-07-23 06:00:00,085] ERROR [KafkaApi-1] Error processing ProducerRequest with correlation id 1647844 from client samza_producer-analytics_2-1-1437158919583-2 on partition [input,3] (kafka.server.KafkaApis)
kafka.message.InvalidMessageException: Message found with corrupt size (0)
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNextOuter(ByteBufferMessageSet.scala:159)
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:192)
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:146)
        at kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:66)
        at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:58)
        at kafka.message.ByteBufferMessageSet$$anon$1.innerDone(ByteBufferMessageSet.scala:150)
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:191)
        at kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:146)
        at kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:66)
        at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:58)
        at scala.collection.Iterator$$anon$19.hasNext(Iterator.scala:400)
        at scala.collection.Iterator$class.foreach(Iterator.scala:772)
        at scala.collection.Iterator$$anon$19.foreach(Iterator.scala:399)
        at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
        at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:102)
        at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:250)
        at scala.collection.Iterator$$anon$19.toBuffer(Iterator.scala:399)
        at kafka.message.ByteBufferMessageSet.assignOffsets(ByteBufferMessageSet.scala:220)
        at kafka.log.Log.liftedTree1$1(Log.scala:275)
        at kafka.log.Log.append(Log.scala:274)
        at kafka.cluster.Partition$$anonfun$appendMessagesToLeader$1.apply(Partition.scala:379)
        at kafka.cluster.Partition$$anonfun$appendMessagesToLeader$1.apply(Partition.scala:365)
        at kafka.utils.Utils$.inLock(Utils.scala:535)
        at kafka.utils.Utils$.inReadLock(Utils.scala:541)
        at kafka.cluster.Partition.appendMessagesToLeader(Partition.scala:365)
        at kafka.server.KafkaApis$$anonfun$appendToLocalLog$2.apply(KafkaApis.scala:291)
        at kafka.server.KafkaApis$$anonfun$appendToLocalLog$2.apply(KafkaApis.scala:282)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
        at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:95)
        at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:95)
        at scala.collection.Iterator$class.foreach(Iterator.scala:772)
        at scala.collection.mutable.HashTable$$anon$1.foreach(HashTable.scala:157)
        at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:190)
        at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:45)
        at scala.collection.mutable.HashMap.foreach(HashMap.scala:95)
        at scala.collection.TraversableLike$class.map(TraversableLike.scala:233)
        at scala.collection.mutable.HashMap.map(HashMap.scala:45)
        at kafka.server.KafkaApis.appendToLocalLog(KafkaApis.scala:282)
        at kafka.server.KafkaApis.handleProducerOrOffsetCommitRequest(KafkaApis.scala:204)
        at kafka.server.KafkaApis.handle(KafkaApis.scala:59)
        at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:59)
        at java.lang.Thread.run(Thread.java:745)