[Spark] Let cancel() wait for the execution thread before stopping the session - #40098
Closed
tkaymak wants to merge 1 commit into
Closed
[Spark] Let cancel() wait for the execution thread before stopping the session#40098tkaymak wants to merge 1 commit into
tkaymak wants to merge 1 commit into
Conversation
…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.
Contributor
Author
|
Folded into #40090, the failure it fixes was found by that PR's CI run and the review there covers it. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
SparkStructuredStreamingPipelineResult.cancel()interrupts the execution thread withFuture.cancel(true)and then runs the terminal state callback at once, which stops the SparkSession unlessuseActiveSparkSessionis set. The interrupt does not wait for the thread. It keeps translating or evaluating on a stopped SparkContext and dies withIllegalStateException: 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
PipelineTranslatorBatchon the execution thread. The class is then poisoned for the JVM and every later batch pipeline fails withNoClassDefFoundError: Could not initialize class PipelineTranslatorBatch. That is the failure ofbeam_PreCommit_Java_Spark_Versionson #40090 (run 34484375079, 12 batch tests in one fork), the first run of the Spark 4 module since #40093 restored it.StructuredStreamingPipelineStateTestcancels right afterrun()in three of its tests, and with a diagnostic log on the execution thread all three show the stopped context locally.Fix:
runAsynchands the result its single thread executor, already shut down after submit, andcancel()waits for it to terminate, bounded at 60 seconds, before the terminal state callback runs. Streaming pipelines end the thread throughStreamingEvaluationContext.stop(), batch pipelines through the interrupt, so the bound is not reached in normal use. NewSparkStructuredStreamingPipelineResultTestchecks that the callback runs only after the execution thread has ended.