[
https://issues.apache.org/jira/browse/SPARK-17147?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16338032#comment-16338032
]
Justin Miller commented on SPARK-17147:
---------------------------------------
I'm also seeing this behavior on a topic that has cleanup.policy=delete. The
volume on this topic is very large, > 10 billion messages per day, and it seems
to happen about once per day. Another topic with lower volume but larger
messages happens every few days.
18/01/23 18:30:10 WARN TaskSetManager: Lost task 28.0 in stage 26.0 (TID 861,
<ip>,executor 15): java.lang.AssertionError: assertion failed: Got wrong record
for <consumergroup>-8002 <topic> 124 even after seeking to offset 1769485661
got back record.offset 1769485662
18/01/23 18:30:12 INFO TaskSetManager: Lost task 28.1 in stage 26.0 (TID 865)
on <ip>,executor 24: java.lang.AssertionError (assertion failed: Got wrong
record for <consumergroup>-8002 <topic> 124 even after seeking to offset
1769485661 got back record.offset 1769485662) [duplicate 1]
18/01/23 18:30:14 INFO TaskSetManager: Lost task 28.2 in stage 26.0 (TID 866)
on <ip>,executor 15: java.lang.AssertionError (assertion failed: Got wrong
record for <consumergroup>-8002 <topic> 124 even after seeking to offset
1769485661 got back record.offset 1769485662) [duplicate 2]
18/01/23 18:30:15 INFO TaskSetManager: Lost task 28.3 in stage 26.0 (TID 867)
on <ip>,executor 15: java.lang.AssertionError (assertion failed: Got wrong
record for <consumergroup>-8002 <topic> 124 even after seeking to offset
1769485661 got back record.offset 1769485662) [duplicate 3]
18/01/23 18:30:18 WARN TaskSetManager: Lost task 28.0 in stage 27.0 (TID 898,
<ip>,executor 6): java.lang.AssertionError: assertion failed: Got wrong record
for <consumergroup>-8002 <topic> 124 even after seeking to offset 1769485661
got back record.offset 1769485662
18/01/23 18:30:19 INFO TaskSetManager: Lost task 28.1 in stage 27.0 (TID 900)
on <ip>,executor 15: java.lang.AssertionError (assertion failed: Got wrong
record for <consumergroup>-8002 <topic> 124 even after seeking to offset
1769485661 got back record.offset 1769485662) [duplicate 1]
18/01/23 18:30:20 INFO TaskSetManager: Lost task 28.2 in stage 27.0 (TID 901)
on <ip>,executor 15: java.lang.AssertionError (assertion failed: Got wrong
record for <consumergroup>-8002 <topic> 124 even after seeking to offset
1769485661 got back record.offset 1769485662) [duplicate 2]
18/01/23 18:30:21 INFO TaskSetManager: Lost task 28.3 in stage 27.0 (TID 902)
on <ip>,executor 15: java.lang.AssertionError (assertion failed: Got wrong
record for <consumergroup>-8002 <topic> 124 even after seeking to offset
1769485661 got back record.offset 1769485662) [duplicate 3]
When checked with kafka-simple-consumer-shell, the offset is in fact missing:
next offset = 1769485661
next offset = 1769485663
next offset = 1769485664
next offset = 1769485665
I'm currently testing out this branch in the persister and will post if it
crashes again over the next few days (I currently have the kafka-10 source from
the branch with a few extra log lines deployed). We're currently on log format
0.10.2 (upgraded yesterday) but saw the same issue on 0.9.0.0.
chao.wu - Is this behavior similar to what you're seeing?
> Spark Streaming Kafka 0.10 Consumer Can't Handle Non-consecutive Offsets
> (i.e. Log Compaction)
> ----------------------------------------------------------------------------------------------
>
> Key: SPARK-17147
> URL: https://issues.apache.org/jira/browse/SPARK-17147
> Project: Spark
> Issue Type: Bug
> Components: DStreams
> Affects Versions: 2.0.0
> Reporter: Robert Conrad
> Priority: Major
>
> When Kafka does log compaction offsets often end up with gaps, meaning the
> next requested offset will be frequently not be offset+1. The logic in
> KafkaRDD & CachedKafkaConsumer has a baked in assumption that the next offset
> will always be just an increment of 1 above the previous offset.
> I have worked around this problem by changing CachedKafkaConsumer to use the
> returned record's offset, from:
> {{nextOffset = offset + 1}}
> to:
> {{nextOffset = record.offset + 1}}
> and changed KafkaRDD from:
> {{requestOffset += 1}}
> to:
> {{requestOffset = r.offset() + 1}}
> (I also had to change some assert logic in CachedKafkaConsumer).
> There's a strong possibility that I have misconstrued how to use the
> streaming kafka consumer, and I'm happy to close this out if that's the case.
> If, however, it is supposed to support non-consecutive offsets (e.g. due to
> log compaction) I am also happy to contribute a PR.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
