[
https://issues.apache.org/jira/browse/SPARK-18057?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16441485#comment-16441485
]
Jordan Moore commented on SPARK-18057:
--------------------------------------
Hi all, chiming in here to point out a production issue we are currently
seeing.
We recently upgraded from Confluent 3.1.2 (Kafka 0.10.1.1) to Confluent 3.3.1
(Kafka 0.11.0.1), and seeing messages such as
{code:java}
Tried to fetch 473151075 but the returned record offset was 473151072{code}
Full Stacktrace below
So, looking into the raw topic with a 0.10.x Java consumer, we see that there
are duplicated offsets (see those ending in 72-74 below), however when we
deserialize the messages, the record values are actually different.
{code:java}
offset = 473151070, timestamp=1523743598312, data=[B@5ae9a829
offset = 473151071, timestamp=1523743598211, data=[B@6d8a00e3
offset = 473151072, timestamp=1523743598213, data=[B@548b7f67
offset = 473151073, timestamp=1523743598215, data=[B@7ac7a4e4
offset = 473151074, timestamp=1523743598423, data=[B@6d78f375
offset = 473151072, timestamp=1523743598831, data=[B@50c87b21
offset = 473151073, timestamp=1523743598837, data=[B@5f375618
offset = 473151074, timestamp=1523743599017, data=[B@1810399e
offset = 473151075, timestamp=1523743599020, data=[B@32d992b2
offset = 473151076, timestamp=1523743599710, data=[B@215be6bb
offset = 473151077, timestamp=1523743599714, data=[B@4439f31e{code}
Running the same simple consumer with at least Kafka 0.11.x libraries *fixed*
the issue.
So, my question here is - even if there isn't a big push towards jumping all
the way onto 1.1.0, what about simply upgrading to at least 0.11? Or how safe
is it to just add {{org.apache.kafka:kafka-clients:0.11.0.0 }}?
Full Stacktrace...
{code:java}
GScheduler: ResultStage 0 (start at SparkStreamingTask.java:222) failed in
77.546 s due to Job aborted due to stage failure: Task 86 in stage 0.0 failed 4
times, most recent failure: Lost task 86.3 in stage 0.0 (TID 96,
ip-10-120-12-52.ec2.internal, executor 11): java.lang.IllegalStateException:
Tried to fetch 473151075 but the returned record offset was 473151072
at
org.apache.spark.sql.kafka010.CachedKafkaConsumer.fetchData(CachedKafkaConsumer.scala:234)
at
org.apache.spark.sql.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:106)
at
org.apache.spark.sql.kafka010.KafkaSourceRDD$$anon$1.getNext(KafkaSourceRDD.scala:158)
at
org.apache.spark.sql.kafka010.KafkaSourceRDD$$anon$1.getNext(KafkaSourceRDD.scala:149)
at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
at scala.collection.convert.Wrappers$IteratorWrapper.hasNext(Wrappers.scala:30)
at
com.domain.spark.KafkaConnectorTask.lambda$run$97bfadb$1(KafkaConnectorTask.java:81)
at org.apache.spark.sql.Dataset$$anonfun$48.apply(Dataset.scala:2269)
at org.apache.spark.sql.Dataset$$anonfun$48.apply(Dataset.scala:2269)
at
org.apache.spark.sql.execution.MapPartitionsExec$$anonfun$6.apply(objects.scala:196)
at
org.apache.spark.sql.execution.MapPartitionsExec$$anonfun$6.apply(objects.scala:193)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:108)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748){code}
> Update structured streaming kafka from 0.10.0.1 to 1.1.0
> --------------------------------------------------------
>
> Key: SPARK-18057
> URL: https://issues.apache.org/jira/browse/SPARK-18057
> Project: Spark
> Issue Type: Improvement
> Components: Structured Streaming
> Reporter: Cody Koeninger
> Priority: Major
>
> There are a couple of relevant KIPs here,
> https://archive.apache.org/dist/kafka/0.10.1.0/RELEASE_NOTES.html
--
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
