Ilya Vasilyev created SPARK-59662:
-------------------------------------
Summary: 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
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]
