[jira] [Created] (SPARK-51083) JavaUtils swallow interrupted exceptions and can cause executor loss

["Neil Ramaswamy (Jira)" <ji...@apache.org>]

Neil Ramaswamy created SPARK-51083:
--------------------------------------

             Summary: JavaUtils swallow interrupted exceptions and can cause executor loss
                 Key: SPARK-51083
                 URL: https://issues.apache.org/jira/browse/SPARK-51083
             Project: Spark
          Issue Type: Improvement
          Components: Spark Core
    Affects Versions: 4.0.0
            Reporter: Neil Ramaswamy


[This line|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java#L184] of JavaUtils.java swallows InterruptedExceptions thrown by [this line|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java#L183], which can cause executor loss.

Consider a streaming task running a streaming aggregation. It takes the following steps:
 # It [writes|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala#L720] all data to the state store, and then [reads|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala#L732] this data back to emit it (in, say, Complete mode).
 # With the RocksDB state store, it doesn't actually acquire any locks, perform any sleeping, etc. during (1). This means that if the query is cancelled and it is interrupted, it won't respond to the interrupt.
 # After (1) is done, it [calls commit|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala#L744] on the state store.
 # In RocksDB, this turns into a call to `RocksDB::commit`, which can call [createSnapshot|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala#L879].
 # In createSnapshot, [we call Utils.deleteRecursively|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala#L879] to make sure that the temporary directory into which we place the RocksDB checkpoint is empty.   
 # In deleteRecursively, we [call deleteRecursivelyUsingUnixNative|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java#L115] which issues an `rm -rf` and then [calls process.waitFor|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java#L183], which declares that it _could_ throw an InterruptedException.
 # Unfortunately, deleteRecursivelyUsingUnixNative [indiscriminately catches all Exceptions|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java#L184], and rethrows an IOException.
 # That IOException is then caught [here|https://github.com/apache/spark/blob/1fc9d7d92ec60cc21e8fd54562e702ea67cf01d3/common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java#L117], and a warning is logged. The interrupt has now been swallowed.

A streaming task thread that misses this interrupted exception will now not exit. If it doesn't exit for 60 seconds, then the TaskReaper will kill the JVM.

Here's my explanation for why I think that we haven't seen this more: if a streaming task thread is able to finish up within 60 seconds of the query cancellation, then the task reaper will not kill it, even if it missed the interrupt. But if it's not able to finish within 60 seconds _and_ it misses the interrupt, you get an executor loss. In a table:
||Finish within 60 seconds of query cancellation?||Receive interrupt within 60 seconds of query cancellation?||Outcome||
|Yes|Yes|No executor loss :)|
|Yes|No|No executor loss :)|
|No|Yes|No executor loss :)|
|No|No|Executor loss :(|

This ticket is about the last row in the table. We should not swallow the exception in `JavaUtils` so that we can respond to the interrupt.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org