[
https://issues.apache.org/jira/browse/SPARK-59662?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18117124#comment-18117124
]
Ilya Vasilyev commented on SPARK-59662:
---------------------------------------
Attached a python script which recreates the described condition and causes
spark to crash.
Would not let me attach it when creating the issue for some reason
> Check for isRepartitionBatch can crash if metadata keys are missing
> -------------------------------------------------------------------
>
> Key: SPARK-59662
> URL: https://issues.apache.org/jira/browse/SPARK-59662
> Project: Spark
> Issue Type: Bug
> Components: SQL, Structured Streaming
> Affects Versions: 4.2.0
> Reporter: Ilya Vasilyev
> Priority: Major
> Attachments: spark-crash.py
>
>
> Hello,
> Ran into an issue where the combination of:
> {code:java}
> spark.sql.streaming.offsetLog.formatVersion=v2
> spark.sql.streaming.checkUnfinishedRepartitionOnRestart=true (the default in
> 4.2.0){code}
> can cause spark to crash if a started batch did not commit:
>
> {code:java}
> 26/09/19 14:05:17 ERROR MicroBatchExecution: Query [id =
> 41da1d72-c034-4a36-9ae3-fdc0e1297405, runId =
> 4eb3a723-257b-45c7-a2e0-72dff79ebf5b] terminated with error
> java.util.NoSuchElementException: None.get
> at scala.None$.get(Option.scala:627)
> at scala.None$.get(Option.scala:626)
> at
> org.apache.spark.sql.execution.streaming.state.OfflineStateRepartitionUtils$.isRepartitionBatch(OfflineStateRepartitionUtils.scala:54)
> at
> org.apache.spark.sql.execution.streaming.runtime.MicroBatchExecution.checkUnfinishedRepartitionBatch(MicroBatchExecution.scala:564)
> at
> org.apache.spark.sql.execution.streaming.runtime.MicroBatchExecution.initializeExecution(MicroBatchExecution.scala:494)
> at
> org.apache.spark.sql.execution.streaming.runtime.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:604)
> at
> org.apache.spark.sql.execution.streaming.runtime.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:353)
> at
> scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
> at
> org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:810)
> at
> org.apache.spark.sql.execution.streaming.runtime.StreamExecution.org$apache$spark$sql$execution$streaming$runtime$StreamExecution$$runStream(StreamExecution.scala:313)
> at
> org.apache.spark.sql.execution.streaming.runtime.StreamExecution$$anon$1.run(StreamExecution.scala:236)
> Traceback (most recent call last): {code}
> The checks in OfflineStateRepartitionUtils.isRepartitionBatch directly access
> the metadata shuffle partitions with .get without checking, but using the v2
> offset log format I believe they are not guaranteed to always exist.
>
> This behavior is was added (at least in part) by combined logic of
> [https://github.com/apache/spark/pull/52645] and
> [https://github.com/apache/spark/pull/53881] and perhaps others
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
