Please see that driver side for example resolved in 3.1.0... G
On Fri, Mar 12, 2021 at 1:03 PM Sachit Murarka <connectsac...@gmail.com> wrote: > Hi Gabor, > > Thanks a lot for the response. I am using Spark 3.0.1 and this is spark > structured streaming. > > Kind Regards, > Sachit Murarka > > > On Fri, Mar 12, 2021 at 5:30 PM Gabor Somogyi <gabor.g.somo...@gmail.com> > wrote: > >> Since you've not provided any version I guess you're using 2.x and you're >> hitting this issue: https://issues.apache.org/jira/browse/SPARK-28367 >> The executor side must be resolved out of the box in the latest Spark >> version however on driver side one must set " >> spark.sql.streaming.kafka.useDeprecatedOffsetFetching=false" to use the >> new way of fetching. >> >> If it doesn't solve your problem then Kafka side must be checked why it's >> not returning... >> >> Hope this helps! >> >> G >> >> >> On Fri, Mar 12, 2021 at 12:29 PM Sachit Murarka <connectsac...@gmail.com> >> wrote: >> >>> Hi All, >>> >>> I am getting following error in spark structured streaming while >>> connecting to Kakfa >>> >>> Main issue from logs:: >>> Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout of >>> 60000ms expired before the position for partition my-topic-1 could be >>> determined >>> >>> Current Committed Offsets: {KafkaV2[Subscribe[my-topic]]: >>> {“my-topic”:{“1":1498,“0”:1410}}} >>> Current Available Offsets: {KafkaV2[Subscribe[my-topic]]: >>> {“my-topic”:{“1”:1499,“0":1410}}} >>> >>> >>> Full logs:: >>> >>> 21/03/12 11:04:35 ERROR TaskSetManager: Task 0 in stage 0.0 failed 4 >>> times; aborting job >>> 21/03/12 11:04:35 ERROR WriteToDataSourceV2Exec: Data source write >>> support >>> org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@1eff441c >>> is aborting. >>> 21/03/12 11:04:35 ERROR WriteToDataSourceV2Exec: Data source write >>> support >>> org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@1eff441c >>> aborted. >>> 21/03/12 11:04:35 ERROR MicroBatchExecution: Query [id = >>> 2d788a3a-f0ee-4903-9679-0d13bc401e12, runId = >>> 1b387c28-c8e3-4336-9c9f-57db16aa8132] terminated with error >>> org.apache.spark.SparkException: Writing job aborted. >>> at >>> org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2(WriteToDataSourceV2Exec.scala:413) >>> at >>> org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2$(WriteToDataSourceV2Exec.scala:361) >>> at >>> org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec.writeWithV2(WriteToDataSourceV2Exec.scala:322) >>> at >>> org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec.run(WriteToDataSourceV2Exec.scala:329) >>> at >>> org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result$lzycompute(V2CommandExec.scala:39) >>> at >>> org.apache.spark.sql.execution.datasources.v2.V2CommandExec.result(V2CommandExec.scala:39) >>> at >>> org.apache.spark.sql.execution.datasources.v2.V2CommandExec.executeCollect(V2CommandExec.scala:45) >>> at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3627) >>> at org.apache.spark.sql.Dataset.$anonfun$collect$1(Dataset.scala:2940) >>> at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3618) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87) >>> at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:764) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64) >>> at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3616) >>> at org.apache.spark.sql.Dataset.collect(Dataset.scala:2940) >>> at >>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$16(MicroBatchExecution.scala:575) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87) >>> at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:764) >>> at >>> org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64) >>> at >>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$15(MicroBatchExecution.scala:570) >>> at >>> org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:352) >>> at >>> org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:350) >>> at >>> org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:69) >>> at >>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.runBatch(MicroBatchExecution.scala:570) >>> at >>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:223) >>> at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) >>> at >>> org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:352) >>> at >>> org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:350) >>> at >>> org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:69) >>> at >>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:191) >>> at >>> org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:57) >>> at >>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:185) >>> at org.apache.spark.sql.execution.streaming.StreamExecution.org >>> $apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:334) >>> at >>> org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:245) >>> Caused by: org.apache.spark.SparkException: Job aborted due to stage >>> failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task >>> 0.3 in stage 0.0 (TID 3, 10.244.2.68, executor 1): >>> org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired >>> before the position for partition my-topic-1 could be determined >>> >>> Driver stacktrace: >>> at >>> org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2059) >>> at >>> org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2008) >>> at >>> org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2007) >>> at >>> scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62) >>> at >>> scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55) >>> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) >>> at >>> org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2007) >>> at >>> org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:973) >>> at >>> org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:973) >>> at scala.Option.foreach(Option.scala:407) >>> at >>> org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:973) >>> at >>> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2239) >>> at >>> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2188) >>> at >>> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2177) >>> at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49) >>> at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:775) >>> at org.apache.spark.SparkContext.runJob(SparkContext.scala:2099) >>> at >>> org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.writeWithV2(WriteToDataSourceV2Exec.scala:382) >>> ... 37 more >>> Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout of >>> 60000ms expired before the position for partition my-topic-1 could be >>> determined >>> >>> Current Committed Offsets: {KafkaV2[Subscribe[my-topic]]: >>> {“my-topic”:{“1":1498,“0”:1410}}} >>> Current Available Offsets: {KafkaV2[Subscribe[my-topic]]: >>> {“my-topic”:{“1”:1499,“0":1410}}} >>> >>> Kind Regards, >>> Sachit Murarka >>> >>
