[Spark][#36841] Translate stateless streaming pipelines on the Spark 4 runner

[tkaymak]

Part of #36841, follows #39971 (DataSourceV2 unbounded source). This makes that source reachable: the Spark 4 runner now translates and runs stateless streaming pipelines.

Scope

Supported in streaming mode: Read.from(UnboundedSource), stateless single output ParDo without side inputs, Window.Assign, Flatten, Reshuffle. The last four reuse the batch translators unchanged. Everything else fails at translation with an UnsupportedOperationException pointing at #36841: GroupByKey, Combine.perKey, stateful ParDo, ParDo with side inputs or additional outputs, Impulse and bounded reads (so Create and PAssert). The rejections are explicit because the batch translators for those primitives persist or collect the Dataset, which Spark refuses on a streaming plan with a raw AnalysisException. GroupByKey and stateful ParDo arrive with the transformWithState bridge in the next PR.

Main code, four files under runners/spark/4

• translation/PipelineTranslatorFactory.java shadows the shared base file that throws for streaming today. The Spark 4 module compiles the override tree with later wins, so only Spark 4 gets the streaming dispatch.
• translation/PipelineTranslatorStreaming.java routes PrimitiveUnboundedRead to the new translator, rejects the unsupported primitives, and falls back to the common registry for the rest. Without the rejections GroupByKey would silently run the batch translator against a streaming Dataset.
• translation/StreamingEvaluationContext.java starts one noop sink query per leaf with a processing time trigger of maxBatchDurationMillis, checkpoints under checkpointDir/<leaf index>, blocks until every query terminates, stops siblings when one fails, and stops a query after streamingStopAfterIdleBatches triggers without input when that option is set. Idle triggers arrive as zero row progress events while the source offset moves and as QueryIdleEvent otherwise, the listener counts both. checkpointDir must be set, the shared default is /tmp/<jobName>.
• translation/streaming/ReadUnboundedTranslator.java builds the Dataset through UnboundedSourceDataset.of and decodes the payload column with the full windowed value coder.

Shared code, runners/spark/src

• The translator registry moved from PipelineTranslatorBatch to PipelineTranslatorCommon, PipelineTranslatorBatch is a thin subclass and the streaming translator extends the common class, so batch only translators added later do not reach the streaming path by inheritance.
• SparkStructuredStreamingPipelineResult.cancel() waits for the execution thread before the terminal state callback stops the SparkSession. Before, the thread kept translating on a stopped SparkContext, and when that hit the cold static init of PipelineTranslatorBatch the class was poisoned for the JVM, which is what the first Spark Versions run on this PR showed. Covered by SparkStructuredStreamingPipelineResultTest.

Tests, all live StreamingQuery runs

• TestUnboundedSource is the source that BeamMicroBatchSourceTest used as a nested class in [Spark][#36841] Add the DataSourceV2 unbounded source for the Spark 4 streaming runner #39971, extracted so the translator tests share it. No second synthetic source.
• StatelessParDoStreamingTest: pass through, and Flatten of two unbounded reads.
• StreamingPipelineLifecycleTest: RUNNING to DONE on idle, cancel, a failing leaf fails the pipeline and stops its healthy sibling.
• StreamingCheckpointRestartTest: two runs against one checkpoint location with the reader cache wiped in between, the second run recreates readers from the durable marks and re-emits nothing the first run committed. The file layout itself is covered by BeamMicroBatchSourceTest, here only the wiring of the checkpoint location is asserted.
• PipelineTranslatorStreamingTest: the rejections surface from run() with the Beam message, not a Spark one.

Delivery is at least once, as documented on BeamReaderCache in #39971. No CHANGES.md entry yet, that comes when the runner can execute a windowed GroupByKey.

R: @Abacn

[github-actions]

Assigning reviewers:

R: @chamikaramj added as fallback since no labels match configuration

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[tkaymak]

Waiting for #40093 to be merged, then the Spark Precommit can run here

[Abacn]

Thanks. Close and reopen PR to triggger tests

[Abacn]

Thanks, had a few comments.

[tkaymak]

Thanks you!
All ten addressed in 58303d6. The registry split is PipelineTranslatorCommon with a thin PipelineTranslatorBatch, as suggested.

One more commit is folded in, 75dbe36: the red Spark Versions run on this PR was cancel() stopping the SparkSession while the execution thread was still translating. In that fork it hit the cold static init of PipelineTranslatorBatch and poisoned the JVM tests.
cancel() now waits for the execution thread before the session stop, with a unit test.
It was pre-existing in the shared runner, it surfaced because #40093 made the module run in CI again.

[Abacn]

the red Spark Versions run on this PR was cancel() stopping the SparkSession while the execution thread was still translating.
It was pre-existing in the shared runner

This could be a real bug. Is this a test timing or an issue could affect production? add 60s timeout is just an ad-hoc mitigation and is non-deterministic

If multiple tests / pipeline share same Spark session, cancel the session should only happen after all tests completed.

We can scope out the long term fix as separate PR though. But should be a known blocker for the Spark 4 runner

[tkaymak]

You are right on both points, it is a production bug and the 60 s wait was a mitigation.
I reverted it here, this PR is the translator slice again, and filed #40101 as the blocker.

What is established, from the Spark 4.0.2 sources:

• cancel() interrupts the execution thread. An interrupt does not cancel a Spark job, DAGScheduler.runJob waits on the JobWaiter until it is cancelled explicitly. Today only the session stop ends batch work, so with useActiveSparkSession=true a batch pipeline is never cancelled and cancel() still reports CANCELLED.
• SparkSessionFactory.getOrCreateSession uses Builder.getOrCreate, which adopts a usable default session, so two pipelines in one JVM share it and cancelling one stops the SparkContext under the other. The legacy runner has ownership guards, this one has none.
• Reproduced locally: after cancel() the execution thread keeps translating and fails with Cannot call methods on a stopped SparkContext. Why that turned into the PipelineTranslatorBatch class init failure in CI is not proven, the primary trace was swallowed by a test that never reads its result. Nothing in that static chain calls the SparkContext.

Suggested Fix (in a separate PR today): job group cancellation in EvaluationContext, session ownership with a reference count in SparkSessionFactory, session stop on the execution thread after evaluation, and an unbounded join in cancel() without Future.cancel(true). Until it lands this PR can hit the race in the Spark Versions job, one failure in three runs so far.

[tkaymak]

The cancel fix is #40103, fixes #40101.

[Abacn]

Thanks!