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
>>>>>
>>>>
>>>>
>>>
>>
>
