tkaymak opened a new issue, #40101: URL: https://github.com/apache/beam/issues/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 #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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
