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