You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
[Bug][Spark] SparkStructuredStreamingRunner cancel() does not cancel Spark jobs and stops a SparkSession it may not own #40101
SparkStructuredStreamingPipelineResult.cancel() interrupts the execution thread, which does not cancel a Spark job (DAGScheduler.runJob only ends through JobWaiter.cancel), then stops the SparkSession even when SparkSession.Builder.getOrCreate returned a pre-existing default session shared with other pipelines in the JVM. Batch EvaluationContext.stop() is a no-op.
Consequences:
With useActiveSparkSession=true a batch pipeline is never cancelled and cancel() still returns CANCELLED.
Two pipelines sharing the default session: cancelling one stops the SparkContext under the other.
The execution thread keeps translating on a stopped SparkContext. In beam_PreCommit_Java_Spark_Versions on [Spark][#36841] Translate stateless streaming pipelines on the Spark 4 runner #40090 this surfaced as a NoClassDefFoundError: Could not initialize class PipelineTranslatorBatch cascade over 12 batch tests in one fork, the primary trace was not captured.
Fix: job group cancellation in EvaluationContext (setJobGroup on the execution thread, cancelJobGroup in stop()), session ownership with a reference count in SparkSessionFactory, session stop on the execution thread after evaluation, unbounded join in cancel() instead of Future.cancel(true).
Known blocker for declaring the Spark 4 Structured Streaming runner stable, #36841. Found in review of #40090.
Activity
[Spark] Make cancel() cancel the Spark jobs and stop only a session t…
[Spark] Make cancel() cancel the Spark jobs and stop only a session t…
[Spark] Make cancel() cancel the Spark jobs and stop only a session t…
[Spark] Make cancel() cancel the Spark jobs and stop only a session t…
[Spark] Make cancel() cancel the Spark jobs and stop only a session t…
[Spark] Make cancel() cancel the Spark jobs and stop only a session t…