[SPARK-25158][SQL]Executor accidentally exit because ScriptTransformationWriterThread throw Exception. #22149

LuciferYang · 2018-08-20T06:12:28Z

What changes were proposed in this pull request?

Run Spark-Sql job use transform features(ScriptTransformationExec) with config spark.speculation = true, sometimes job fails and we found many Executor Dead through Executor Tab, through analysis log and code we found :

ScriptTransformationExec start a new thread(ScriptTransformationWriterThread), the new thread is very likely to throw TaskKilledException(from iter.map.foreach part) when speculation is on, this exception will captured by SparkUncaughtExceptionHandler which registered during Executor start, SparkUncaughtExceptionHandler will call System.exit (SparkExitCode.UNCAUGHT_EXCEPTION) to shutdown Executor, this is unexpected.

We should not kill the executor just because ScriptTransformationWriterThread fails. log the error(not only TaskKilledException) instead of throwing it is enough, Exception already pass to ScriptTransformationExec and handle by TaskRunner.

How was this patch tested?

Register TestUncaughtExceptionHandler to test case in ScriptTransformationSuite, then assert there is no Uncaught Exception handled.

Before this patch "script transformation should not swallow errors from upstream operators (no serde)" and "script transformation should not swallow errors from upstream operators (with serde)" throwing IllegalArgumentException and handle by TestUncaughtExceptionHandler .

…kKilledException

LuciferYang · 2018-08-20T09:33:51Z

ping @xuanyuanking

xuanyuanking

LGTM, we have tested this in our product env. cc @cloud-fan for the further review.

xuanyuanking · 2018-08-23T11:15:00Z

Gental ping @gatorsmile.

gatorsmile · 2018-08-27T23:29:35Z

Is that possible to add a test case?

gatorsmile · 2018-08-27T23:29:54Z

ok to test

SparkQA · 2018-08-28T01:37:49Z

Test build #95320 has finished for PR 22149 at commit 412497f.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

xuanyuanking · 2018-08-28T02:13:26Z

retest this please.

xuanyuanking · 2018-08-28T02:21:20Z

Is that possible to add a test case?

Thanks for your reply Xiao, we encountered some difficulties during the test case, cause this need mock on speculative behavior. We will keep looking into this.

SparkQA · 2018-08-28T04:46:20Z

Test build #95323 has finished for PR 22149 at commit 412497f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

LuciferYang · 2018-08-30T04:52:16Z

 Is that possible to add a test case?

Thanks for your reply Xiao, I tried to add a test case to reproduce bad case, but it only reproduce in small probability，we need to ensure that Task still run in iter.map(outputProjection).foreach { row => xxxx} code part of ScriptTransformationWriterThread.run method and not completed when TaskRunner response KillTask cmd , timing is difficult to accuracy control.

LuciferYang · 2018-11-20T07:40:23Z

@gatorsmile should we fix this problem?

cloud-fan · 2018-11-21T02:47:17Z

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/ScriptTransformationExec.scala

@@ -308,6 +308,12 @@ private class ScriptTransformationWriterThread(
      }
      threwException = false
    } catch {
+      // TaskKilledException should not be thrown again, otherwise it will be captured by
+      // SparkUncaughtExceptionHandler, then Executor will exit because of TaskKilledException.
+      case e: TaskKilledException =>


which line of code may throw this exception? and how is this exception handled in other operators?

Thanks for your reply @cloud-fan ,
The error log and problem analysis of one bad case I paste in https://issues.apache.org/jira/browse/SPARK-25158.
In that case .UnsafeInMemorySorter$SortedIterator.loadNext throw TaskKilledException.
In another case TaskKilledException throw by InterruptibleIterator.hasNext , and from the error stack we can see that line 281 iter.map(outputProjection).foreach is the stack root.

Actually ScriptTransformationExec like a streaming pipe, but other operators basically no such characteristics.

On the other hand perhaps ScriptTransformationWriterThread should not throw any Exception because the we do Exception mark and handle by ScriptTransformationExec , but I'm not sure.

Actually ScriptTransformationExec like a streaming pipe, but other operators basically no such characteristics.

can you elaborate on it? How TaskKilledException is handled in other operators?

Sorry, maybe I didn't express it clearly.

There is no essential difference between ScriptTransformationExec and other operators in exception handle, exceptions handle by catch block of Executor.TaskRunner#run method.

The key point isScriptTransformationExec create a new thread named ScriptTransformationWriterThread .

Exception throw from ScriptTransformationWriterThread thread cannot catch by ScriptTransformationExec and catch block of Executor.TaskRunner#run method, then it will handle by SparkUncaughtExceptionHandler which we register in Executor at !isLocal run model.

SparkUncaughtExceptionHandler will call System.exit(SparkExitCode.UNCAUGHT_EXCEPTION) when case _ if exitOnUncaughtException

I see. So ScriptTransformationExec is special because

it starts a new thread

the new thread is very likely to throw TaskKilledException, when speculation is on.

I think we should not kill the executor just because ScriptTransformationWriterThread fails. We should log the error(not only TaskKilledException), instead of throwing it.

Cannot argree more ~

SparkQA · 2018-11-21T10:13:25Z

Test build #99112 has finished for PR 22149 at commit db7167c.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2018-11-21T10:30:30Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/ScriptTransformationSuite.scala

+  def cleanStatus: Unit = _exception = null
+
+  override def uncaughtException(t: Thread, e: Throwable): Unit = {
+    logError(s"Thread ${t.getName} handle by TestUncaughtExceptionHandler")


Do we need to do logging in test?

viirya · 2018-11-21T10:30:49Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/ScriptTransformationSuite.scala

+
+  protected override def beforeAll(): Unit = {
+    super.beforeAll()
+    Thread.setDefaultUncaughtExceptionHandler(uncaughtExceptionHandler)


Shall we better get default handler and restore it?

ok~ getDefaultUncaughtExceptionHandler and store it at beforeAll method, then restore use defaultUncaughtExceptionHandler at afterAll method

viirya · 2018-11-21T10:33:00Z

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/ScriptTransformationExec.scala

      case t: Throwable =>
        // An error occurred while writing input, so kill the child process. According to the
        // Javadoc this call will not throw an exception:
        _exception = t
        proc.destroy()
-        throw t
+        logError(s"Thread-ScriptTransformation-Feed exit cause by: ", t)


nit: no need to use string interpolation?

thx ~ no need~

SparkQA · 2018-11-21T12:38:16Z

Test build #99116 has finished for PR 22149 at commit 6a90e8c.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-11-21T12:47:16Z

Test build #99113 has finished for PR 22149 at commit 077f16a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-11-21T12:56:06Z

Test build #99114 has finished for PR 22149 at commit 82cb013.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-11-21T13:21:05Z

Test build #99115 has finished for PR 22149 at commit 941fd9e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

LuciferYang · 2019-01-20T04:31:59Z

@cloud-fan should we fix this problems?

cloud-fan · 2019-01-21T14:10:34Z

the fix LGTM, but can we provide a better test? IIUC the test can be:

create a UDF which would throw exception when producing data
use spark.range(...).select(myUDF(...)) as input plan and do script transform
assert that the query fails, but the executor is still alive.

LuciferYang · 2019-01-22T05:35:09Z

@cloud-fan ok~

LuciferYang · 2019-02-03T03:39:39Z

@cloud-fan There are two things to confirm before submit the test case：

This case is about hive transform not UDF, so I need add a python script file for this case, will we accept this?
SparkUncaughtExceptionHandler not register to Executor in local mode, the test case maybe only add a TestUncaughtExceptionHandler like I add to ScriptTransformationSuite and register it manually, then verify job fails but there is no exception catch by TestUncaughtExceptionHandler like asserts in ScriptTransformationSuite

LuciferYang · 2019-02-03T03:45:55Z

I found there are some python scripts in spark-hive module, I'll try to reuse it.

SparkQA · 2019-02-03T06:25:49Z

Test build #102019 has finished for PR 22149 at commit 401f7d0.

This patch fails Scala style tests.
This patch does not merge cleanly.
This patch adds the following public classes (experimental):
class TestUncaughtExceptionHandler extends Thread.UncaughtExceptionHandler

SparkQA · 2019-02-03T06:48:14Z

Test build #102020 has finished for PR 22149 at commit 4ed4c9c.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-03T08:05:01Z

Test build #102021 has finished for PR 22149 at commit 33e4343.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2019-02-03T20:03:16Z

retest this please

SparkQA · 2019-02-03T22:33:53Z

Test build #102025 has finished for PR 22149 at commit 33e4343.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2019-02-04T19:24:01Z

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/ScriptTransformationExec.scala

@@ -308,12 +308,15 @@ private class ScriptTransformationWriterThread(
      }
      threwException = false
    } catch {
+      // SPARK-25158 Exception should not be thrown again, otherwise it will be captured by


All exceptions? How about the fatal exceptions? Should we do it only when it is NonFatal?

All exceptions already raise to ScriptTransformationExec use _exception, they're all dealt with catch block of TaskRunner whatever it is a NonFatal Exception or a Fatal Exception

cloud-fan · 2019-02-11T04:14:18Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

+        spark
+          .range(5)
+          .selectExpr("id AS a")
+          .createOrReplaceTempView("test")


wrap the code with withTempView

cloud-fan · 2019-02-11T04:15:30Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

+  test("SPARK-25158: " +
+    "Executor accidentally exit because ScriptTransformationWriterThread throw Exception") {
+    withSQLConf("hive.script.recordwriter" ->
+      classOf[NonWritableRecordWriter].getCanonicalName) {


this works, but I'm wondering if we can replace it with a UDF. e.g.

val badUDF = udf { (id: Long): Long => throw new RuntimeException(...) } spark .range(5) .select(badUDF('id).as("a")) .createOrReplaceTempView("test")

@cloud-fan Now run this case we can see expected log as follwing:

23:32:45.800 ERROR org.apache.spark.sql.hive.execution.ScriptTransformationWriterThread: Thread-ScriptTransformation-Feed exit cause by: org.apache.spark.SparkException: Failed to execute user defined function(SQLQuerySuite$$Lambda$754/1644529474: (int) => int) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(generated.java:106) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:730) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458) at scala.collection.Iterator.foreach(Iterator.scala:941) at scala.collection.Iterator.foreach$(Iterator.scala:941) at scala.collection.AbstractIterator.foreach(Iterator.scala:1429) at org.apache.spark.sql.hive.execution.ScriptTransformationWriterThread.$anonfun$run$1(ScriptTransformationExec.scala:281) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1899) at org.apache.spark.sql.hive.execution.ScriptTransformationWriterThread.run(ScriptTransformationExec.scala:270) Caused by: java.lang.RuntimeException: Failed to produce data. at org.apache.spark.sql.hive.execution.SQLQuerySuite.$anonfun$new$458(SQLQuerySuite.scala:2370) at scala.runtime.java8.JFunction1$mcII$sp.apply(JFunction1$mcII$sp.java:23) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(generated.java:104) ... 10 more

and uncaughtExceptionHandler catch nothing, use the original code uncaughtExceptionHandler will catch a SparkException

SparkQA · 2019-02-11T08:05:01Z

Test build #102179 has finished for PR 22149 at commit 8b30ebc.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-11T08:05:02Z

Test build #102184 has finished for PR 22149 at commit c4887e5.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-02-11T08:44:39Z

retest this please

cloud-fan · 2019-02-11T08:48:22Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

+
+        // Use a bad udf to generate failed inputs.
+        import org.apache.spark.sql.functions.udf
+        val badUDF = udf({x: Int =>


nit:

val badUDF = org.apache.spark.sql.functions.udf((x: Int) => { if ... else throw ... })

cloud-fan · 2019-02-11T08:49:33Z

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

+               |USING 'python $scriptFilePath/scripts/test_transform.py "\t"'
+             """.stripMargin).collect()
+        }
+        assert(e.getMessage.contains("Failed to produce data."))


what's the result if we run this test without your fix and the TestUncaughtExceptionHandler?

Keep throw t, TestUncaughtExceptionHandler will catch a SparkException.
Keep throw t and not set TestUncaughtExceptionHandler as DefaultUncaughtExceptionHandler, this case always success but we can see log as follwing:

ERROR org.apache.spark.util.Utils: Uncaught exception in thread Thread-ScriptTransformation-Feed

and the Uncaught exception will catch by SparkUncaughtExceptionHandler in non local model.

And if keep throw t and use SparkUncaughtExceptionHandler instead of TestUncaughtExceptionHandler , the case always success but we can see log as follwoing:

ERROR org.apache.spark.util.Utils: Uncaught exception in thread Thread-ScriptTransformation-Feed ERROR org.apache.spark.util.SparkUncaughtExceptionHandler: Uncaught exception in thread Thread[Thread-ScriptTransformation-Feed,5,main] Process finished with exit code 50

The test process finished with exit code 50 not 0

The test process finished with exit code 50 not 0

What's the return data of the query then?

No return data... we intercept a SparkException, and the query should failed in local mode, I think we should be concerned that there are no uncaught exceptions like TaskKilledException catch by UncaughtExceptionHandler in non local model .

SparkQA · 2019-02-11T11:06:55Z

Test build #102191 has finished for PR 22149 at commit a5e3d11.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-02-11T11:17:37Z

Test build #102190 has finished for PR 22149 at commit c4887e5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-02-12T04:16:48Z

thanks, merging to master!

LuciferYang · 2019-02-12T05:06:43Z

@cloud-fan thx~

…ationWriterThread throw Exception. ## What changes were proposed in this pull request? Run Spark-Sql job use transform features(`ScriptTransformationExec`) with config `spark.speculation = true`, sometimes job fails and we found many Executor Dead through `Executor Tab`, through analysis log and code we found : `ScriptTransformationExec` start a new thread(`ScriptTransformationWriterThread`), the new thread is very likely to throw `TaskKilledException`(from iter.map.foreach part) when speculation is on, this exception will captured by `SparkUncaughtExceptionHandler` which registered during Executor start, `SparkUncaughtExceptionHandler` will call `System.exit (SparkExitCode.UNCAUGHT_EXCEPTION)` to shutdown `Executor`, this is unexpected. We should not kill the executor just because `ScriptTransformationWriterThread` fails. log the error(not only `TaskKilledException`) instead of throwing it is enough, Exception already pass to `ScriptTransformationExec` and handle by `TaskRunner`. ## How was this patch tested? Register `TestUncaughtExceptionHandler` to test case in `ScriptTransformationSuite`, then assert there is no Uncaught Exception handled. Before this patch "script transformation should not swallow errors from upstream operators (no serde)" and "script transformation should not swallow errors from upstream operators (with serde)" throwing `IllegalArgumentException` and handle by `TestUncaughtExceptionHandler` . Closes apache#22149 from LuciferYang/fix-transformation-task-kill. Authored-by: yangjie01 <yangjie01@baidu.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

fix Executor exit cause by ScriptTransformationWriterThread throw Tas…

412497f

…kKilledException

xuanyuanking approved these changes Aug 20, 2018

View reviewed changes

cloud-fan reviewed Nov 21, 2018

View reviewed changes

use logerror instead of throw exception

db7167c

LuciferYang added 2 commits November 21, 2018 18:15

fix style

077f16a

remove one empty line

82cb013

viirya reviewed Nov 21, 2018

View reviewed changes

LuciferYang added 2 commits November 21, 2018 18:49

fix some nits

941fd9e

get default handler and restore it

6a90e8c

LuciferYang changed the title ~~[SPARK-25158][SQL]Executor accidentally exit because ScriptTransformationWriterThread throws TaskKilledException.~~ [SPARK-25158][SQL]Executor accidentally exit because ScriptTransformationWriterThread throw Exception. Feb 3, 2019

add a sql case

401f7d0

merge upmaster

4ed4c9c

fix format

33e4343

gatorsmile reviewed Feb 4, 2019

View reviewed changes

cloud-fan reviewed Feb 11, 2019

View reviewed changes

LuciferYang added 2 commits February 11, 2019 15:20

add withTempView

8b30ebc

Use a bad udf to generate failed inputs.

c4887e5

cloud-fan reviewed Feb 11, 2019

View reviewed changes

fix badUDF

a5e3d11

cloud-fan closed this in 5864e8e Feb 12, 2019

LuciferYang deleted the fix-transformation-task-kill branch June 10, 2020 08:42

[SPARK-25158][SQL]Executor accidentally exit because ScriptTransformationWriterThread throw Exception. #22149

[SPARK-25158][SQL]Executor accidentally exit because ScriptTransformationWriterThread throw Exception. #22149

Conversation

LuciferYang commented Aug 20, 2018 • edited

What changes were proposed in this pull request?

How was this patch tested?

LuciferYang commented Aug 20, 2018

xuanyuanking left a comment

Choose a reason for hiding this comment

xuanyuanking commented Aug 23, 2018

gatorsmile commented Aug 27, 2018

gatorsmile commented Aug 27, 2018

SparkQA commented Aug 28, 2018

xuanyuanking commented Aug 28, 2018

xuanyuanking commented Aug 28, 2018

SparkQA commented Aug 28, 2018

LuciferYang commented Aug 30, 2018 • edited

LuciferYang commented Nov 20, 2018

Choose a reason for hiding this comment

LuciferYang Nov 21, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang Nov 21, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang Nov 21, 2018 • edited

Choose a reason for hiding this comment

cloud-fan Nov 21, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Nov 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Nov 21, 2018

SparkQA commented Nov 21, 2018

SparkQA commented Nov 21, 2018

SparkQA commented Nov 21, 2018

LuciferYang commented Jan 20, 2019

cloud-fan commented Jan 21, 2019

LuciferYang commented Jan 22, 2019

LuciferYang commented Feb 3, 2019

LuciferYang commented Feb 3, 2019

SparkQA commented Feb 3, 2019

SparkQA commented Feb 3, 2019

SparkQA commented Feb 3, 2019

gatorsmile commented Feb 3, 2019

SparkQA commented Feb 3, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Feb 11, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang Feb 11, 2019 • edited

Choose a reason for hiding this comment

SparkQA commented Feb 11, 2019

SparkQA commented Feb 11, 2019

cloud-fan commented Feb 11, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang Feb 11, 2019 • edited

Choose a reason for hiding this comment

LuciferYang Feb 11, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang Feb 12, 2019 • edited

Choose a reason for hiding this comment

SparkQA commented Feb 11, 2019

SparkQA commented Feb 11, 2019

cloud-fan commented Feb 12, 2019

LuciferYang commented Feb 12, 2019

LuciferYang commented Aug 20, 2018 •

edited

LuciferYang commented Aug 30, 2018 •

edited

LuciferYang Nov 21, 2018 •

edited

LuciferYang Nov 21, 2018 •

edited

LuciferYang Nov 21, 2018 •

edited

cloud-fan Nov 21, 2018 •

edited

cloud-fan Feb 11, 2019 •

edited

LuciferYang Feb 11, 2019 •

edited

LuciferYang Feb 11, 2019 •

edited

LuciferYang Feb 11, 2019 •

edited

LuciferYang Feb 12, 2019 •

edited