spark-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Jordan Moore (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (SPARK-18057) Update structured streaming kafka from 0.10.0.1 to 1.1.0
Date Tue, 17 Apr 2018 20:48:00 GMT

    [ 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-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org


Mime
View raw message