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

[tkaymak]

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.