Hello, Using the old Spark Streaming Kafka API, I got the following around the same offset:
kafka.message.InvalidMessageException: Message is corrupt (stored crc = 3561357254, computed crc = 171652633) at kafka.message.Message.ensureValid(Message.scala:166) at kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:102) at kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:33) at kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:66) at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:58) at org.apache.spark.streaming.kafka.ReliableKafkaReceiver$MessageHandler.run(ReliableKafkaReceiver.scala:265) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) 15/07/20 15:56:57 INFO BlockManager: Removing broadcast 4641 15/07/20 15:56:57 ERROR ReliableKafkaReceiver: Error handling message java.lang.IllegalStateException: Iterator is in failed state at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54) at org.apache.spark.streaming.kafka.ReliableKafkaReceiver$MessageHandler.run(ReliableKafkaReceiver.scala:265) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) I found some old topic about some possible corrupt Kafka message produced by the new producer API with Snappy compression on. My question is, is it possible to skip/ignore those offsets when full processing with KafkaUtils.createStream or KafkaUtils.createDirectStream ? Regards, Nicolas PHUNG On Mon, Jul 20, 2015 at 3:46 PM, Cody Koeninger <c...@koeninger.org> wrote: > I'd try logging the offsets for each message, see where problems start, > then try using the console consumer starting at those offsets and see if > you can reproduce the problem. > > On Mon, Jul 20, 2015 at 2:15 AM, Nicolas Phung <nicolas.ph...@gmail.com> > wrote: > >> Hi Cody, >> >> Thanks for you help. It seems there's something wrong with some messages >> within my Kafka topics then. I don't understand how, I can get bigger or >> incomplete message since I use default configuration to accept only 1Mb >> message in my Kafka topic. If you have any others informations or >> suggestions, please tell me. >> >> Regards, >> Nicolas PHUNG >> >> On Thu, Jul 16, 2015 at 7:08 PM, Cody Koeninger <c...@koeninger.org> >> wrote: >> >>> Not exactly the same issue, but possibly related: >>> >>> https://issues.apache.org/jira/browse/KAFKA-1196 >>> >>> On Thu, Jul 16, 2015 at 12:03 PM, Cody Koeninger <c...@koeninger.org> >>> wrote: >>> >>>> Well, working backwards down the stack trace... >>>> >>>> at java.nio.Buffer.limit(Buffer.java:275) >>>> >>>> That exception gets thrown if the limit is negative or greater than the >>>> buffer's capacity >>>> >>>> >>>> at kafka.message.Message.sliceDelimited(Message.scala:236) >>>> >>>> If size had been negative, it would have just returned null, so we know >>>> the exception got thrown because the size was greater than the buffer's >>>> capacity >>>> >>>> >>>> I haven't seen that before... maybe a corrupted message of some kind? >>>> >>>> If that problem is reproducible, try providing an explicit argument for >>>> messageHandler, with a function that logs the message offset. >>>> >>>> >>>> On Thu, Jul 16, 2015 at 11:28 AM, Nicolas Phung < >>>> nicolas.ph...@gmail.com> wrote: >>>> >>>>> Hello, >>>>> >>>>> When I'm reprocessing the data from kafka (about 40 Gb) with the new >>>>> Spark Streaming Kafka method createDirectStream, everything is fine till >>>>> a driver error happened (driver is killed, connection lost...). When the >>>>> driver pops up again, it resumes the processing with the checkpoint in >>>>> HDFS. Except, I got this: >>>>> >>>>> 15/07/16 15:23:41 ERROR TaskSetManager: Task 4 in stage 4.0 failed 4 >>>>> times; aborting job >>>>> 15/07/16 15:23:41 ERROR JobScheduler: Error running job streaming job >>>>> 1437032118000 ms.0 >>>>> org.apache.spark.SparkException: Job aborted due to stage failure: Task 4 >>>>> in stage 4.0 failed 4 times, most recent failure: Lost task 4.3 in stage >>>>> 4.0 (TID 16, slave05.local): java.lang.IllegalArgumentException >>>>> at java.nio.Buffer.limit(Buffer.java:275) >>>>> at kafka.message.Message.sliceDelimited(Message.scala:236) >>>>> at kafka.message.Message.payload(Message.scala:218) >>>>> at kafka.message.MessageAndMetadata.message(MessageAndMetadata.scala:32) >>>>> at >>>>> org.apache.spark.streaming.kafka.KafkaUtils$$anonfun$6.apply(KafkaUtils.scala:395) >>>>> at >>>>> org.apache.spark.streaming.kafka.KafkaUtils$$anonfun$6.apply(KafkaUtils.scala:395) >>>>> at >>>>> org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.getNext(KafkaRDD.scala:176) >>>>> at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71) >>>>> at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>>> at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>>> at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>>> at >>>>> org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:248) >>>>> at >>>>> org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:172) >>>>> at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:79) >>>>> at org.apache.spark.rdd.RDD.iterator(RDD.scala:242) >>>>> at >>>>> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35) >>>>> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277) >>>>> at org.apache.spark.rdd.RDD.iterator(RDD.scala:244) >>>>> at >>>>> org.apache.spark.rdd.CoalescedRDD$$anonfun$compute$1.apply(CoalescedRDD.scala:93) >>>>> at >>>>> org.apache.spark.rdd.CoalescedRDD$$anonfun$compute$1.apply(CoalescedRDD.scala:92) >>>>> at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371) >>>>> at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>>> at org.elasticsearch.spark.rdd.EsRDDWriter.write(EsRDDWriter.scala:48) >>>>> at >>>>> org.elasticsearch.spark.rdd.EsSpark$$anonfun$saveToEs$1.apply(EsSpark.scala:67) >>>>> at >>>>> org.elasticsearch.spark.rdd.EsSpark$$anonfun$saveToEs$1.apply(EsSpark.scala:67) >>>>> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61) >>>>> at org.apache.spark.scheduler.Task.run(Task.scala:64) >>>>> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203) >>>>> at >>>>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) >>>>> at >>>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) >>>>> at java.lang.Thread.run(Thread.java:745) >>>>> >>>>> This is happening only when I'm doing a full data processing from >>>>> Kafka. If there's no load, when you killed the driver and then restart, it >>>>> resumes the checkpoint as expected without missing data. Did someone >>>>> encounters something similar ? How did you solve this ? >>>>> >>>>> Regards, >>>>> >>>>> Nicolas PHUNG >>>>> >>>> >>>> >>> >> >