Repository navigation
[Spark] Structured streaming runner: asynchronous cancel() should not report CANCELLED before the execution ended #40120
Description
SparkStructuredStreamingPipelineResult.cancel() requests cancellation asynchronously and sets the state to CANCELLED at once. Per the PipelineResult contract the state should be observed: getState() queries the execution, waitUntilFinish() blocks until the pipeline is actually cancelled on the (mini) cluster, and only then the state becomes CANCELLED.
Changing this touches the shared StructuredStreamingPipelineStateTest, which asserts CANCELLED right after cancel() for the batch path today, and the legacy SparkPipelineResult has the same shape. Scope: runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineResult.java and that test.
Activity
tkaymak commentedon Sep 14, 2026
Also part of this:
cancel()called after the execution already ended, but before anyone observed it, currently reports CANCELLED instead of the real DONE or FAILED. The state query should resolve that from the execution.singhmrinal02 commentedon Sep 16, 2026
Hi, I'd like to work on this issue. I'm familiar with Java and I'm willing to learn the relevant Beam/Spark code. Could you please assign this issue to me?
tkaymak commentedon Sep 16, 2026
Hi @singhmrinal02
I am not looking at this exact issue in the next 7 days, as I am afk.
I would continue with the overall effort (#36841), when I am back.
I think it makes sense to coordinate there.
tkaymak commentedon Sep 23, 2026
Hi @singhmrinal02, I am back. Please go ahead, the issue is yours. I am not touching this code for now, my next work on #36841 is the state internals for slice 5.
A few pointers:
SparkStructuredStreamingPipelineResult.cancel()andgetState(). The state should come from the execution, so a cancel after the job already ended reports DONE or FAILED.StructuredStreamingPipelineStateTestasserts CANCELLED right aftercancel(). Those assertions need to move behindwaitUntilFinish().:runners:spark:3:testand:runners:spark:4:test.waitUntilFinish(). Otherwise the session gets stopped under the next test class. This broke the Spark Versions precommit once on [Spark] Make cancel() cancel the Spark jobs and stop only a session the runner created #40103.Abacn reviewed #40103 and will likely review this one. He prefers no blocking waits or extra locks in
cancel().