InvalidMessageSizeException

[Rahul Biswas]

Hi,

We are running into this exception quite frequently now. We are using Storm's 
Kafka Spout as a consumer and as part of a logging message it tries to print 
out the size of the underlying message, when the validation on the size seems 
to fail.

The full stack trace is as below --
kafka.common.InvalidMessageSizeException: invalid message size: 161758511 only 
received bytes: 215831 at 10738556( possible causes (1) a single message larger 
than the fetch size; (2) log corruption )
        at 
kafka.message.ByteBufferMessageSet$$anon$1.makeNextOuter(ByteBufferMessageSet.scala:103)
        at 
kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:138)
        at 
kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:82)
        at 
kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:59)
        at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:51)
        at scala.collection.Iterator$class.foreach(Iterator.scala:631)
        at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:30)
        at scala.collection.IterableLike$class.foreach(IterableLike.scala:79)
        at kafka.message.MessageSet.foreach(MessageSet.scala:87)
        at 
scala.collection.TraversableOnce$class.size(TraversableOnce.scala:100)
        at kafka.message.MessageSet.size(MessageSet.scala:87)
        at storm.kafka.KafkaSpout$PartitionManager.fill(KafkaSpout.java:111)
        at storm.kafka.KafkaSpout$PartitionManager.next(KafkaSpout.java:87)
        at storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:208)
        at backtype.storm.daemon.task$fn__3349$fn__3404.invoke(task.clj:412)
        at 
backtype.storm.daemon.task$mk_task$iter__3281__3285$fn__3286$fn__3287$fn__3288.invoke(task.clj:262)
        at clojure.lang.AFn.applyToHelper(AFn.java:159)
        at clojure.lang.AFn.applyTo(AFn.java:151)
        at clojure.core$apply.invoke(core.clj:601)
        at backtype.storm.util$async_loop$fn__451.invoke(util.clj:369)
        at clojure.lang.AFn.run(AFn.java:24)
        at java.lang.Thread.run(Thread.java:619)
There are also negative message sizes (for a different consumer with a fetch 
size of 5MB) --
kafka.common.InvalidMessageSizeException: invalid message size: -2012627084 
only received bytes: 5242876 at 1054940626667( possible causes (1) a single 
message larger than the fetch size; (2) log core

thanks.
Rahul