apache / apache/beam

[Bug][Spark] SparkStructuredStreamingRunner cancel() does not cancel Spark jobs and stops a SparkSession it may not own

Closed
#40,101 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

`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 #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.

Contributor guide

Open the contributing guide

Research direction

Start by tracing SparkStructuredStreamingPipelineResult.cancel() through EvaluationContext and SparkSessionFactory, including the execution-thread and shared-session paths described in the issue. Run the relevant Spark runner tests, then the beam_PreCommit_Java_Spark_Versions suite; done means job cancellation works, shared sessions remain usable, and cancellation waits for evaluation to finish.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, spark
Domain
data-engineering, distributed-systems, stream-processing
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
55/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.