[SPARK-48717] Python foreachBatch streaming query cannot be stopped gracefully after pin thread mode is enabled and is running spark queries

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 4.0.0
• Fix Version/s: 4.0.0
• Component/s: PySpark, SS
• Labels: pull-request-available
• Assignee: Wei Liu
• Reporter: Wei Liu
• Created: 25/Jun/24 23:58
• Updated: 11/Jul/24 23:40
• Resolved: 27/Jun/24 01:53

Description

Followup of https://issues.apache.org/jira/browse/SPARK-39218

 

It only considered the InterruptedException is thrown when time.sleep(10) is intercepted. But when a spark query is executing:

def func(batch_df, batch_id):
    batch_df.sparkSession.range(10000000).write.saveAsTable("oops")
    print(batch_df.count()) 

the actual error would be:

 

py4j.protocol.Py4JJavaError: An error occurred while calling o2141502.saveAsTable.  
: org.apache.spark.SparkException: Job aborted.  
<REDACTED>
<REDACTED>
<REDACTED>
...
at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$2(StreamExecution.scala:262)  
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)  
at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:262)  
*Caused by: java.lang.InterruptedException  
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1000)*  
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1308)   

We should also add consideration to this scenario 

Issue Links

• links to: GitHub Pull Request #47307