Skip to content

[Spark] Let cancel() wait for the execution thread before stopping the session - #40098

Closed
tkaymak wants to merge 1 commit into
apache:masterfrom
tkaymak:spark-ss-cancel-awaits-execution
Closed

[Spark] Let cancel() wait for the execution thread before stopping the session#40098
tkaymak wants to merge 1 commit into
apache:masterfrom
tkaymak:spark-ss-cancel-awaits-execution

Conversation

@tkaymak

@tkaymak tkaymak commented Sep 11, 2026

Copy link
Copy Markdown
Contributor

SparkStructuredStreamingPipelineResult.cancel() interrupts the execution thread with Future.cancel(true) and then runs the terminal state callback at once, which stops the SparkSession unless useActiveSparkSession is set. The interrupt does not wait for the thread. It keeps translating or evaluating on a stopped SparkContext and dies with IllegalStateException: Cannot call methods on a stopped SparkContext.

When that happens during the first pipeline of a JVM the stop can land inside the static initialization of PipelineTranslatorBatch on the execution thread. The class is then poisoned for the JVM and every later batch pipeline fails with NoClassDefFoundError: Could not initialize class PipelineTranslatorBatch. That is the failure of beam_PreCommit_Java_Spark_Versions on #40090 (run 34484375079, 12 batch tests in one fork), the first run of the Spark 4 module since #40093 restored it. StructuredStreamingPipelineStateTest cancels right after run() in three of its tests, and with a diagnostic log on the execution thread all three show the stopped context locally.

Fix: runAsync hands the result its single thread executor, already shut down after submit, and cancel() waits for it to terminate, bounded at 60 seconds, before the terminal state callback runs. Streaming pipelines end the thread through StreamingEvaluationContext.stop(), batch pipelines through the interrupt, so the bound is not reached in normal use. New SparkStructuredStreamingPipelineResultTest checks that the callback runs only after the execution thread has ended.

…e session

SparkStructuredStreamingPipelineResult.cancel() interrupted the execution
thread and ran the terminal state callback at once, which stops the
SparkSession. The thread kept translating or evaluating on a stopped
SparkContext. When that happened during the static initialization of
PipelineTranslatorBatch the class was poisoned for the JVM and every later
batch pipeline failed with NoClassDefFoundError, seen in the Spark Versions
PreCommit on apache#40090.

runAsync now hands the result its single thread executor and cancel() waits
for it to terminate, bounded at 60 seconds, before the callback runs.
@tkaymak

tkaymak commented Sep 11, 2026

Copy link
Copy Markdown
Contributor Author

Folded into #40090, the failure it fixes was found by that PR's CI run and the review there covers it.

@tkaymak tkaymak closed this Sep 11, 2026
@tkaymak
tkaymak deleted the spark-ss-cancel-awaits-execution branch September 11, 2026 07:43
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant