tkaymak opened a new pull request, #40103: URL: https://github.com/apache/beam/pull/40103
Fixes #40101. Found in review of #40090. **Bug** `SparkStructuredStreamingPipelineResult.cancel()` interrupted the execution thread with `Future.cancel(true)` and then stopped the SparkSession from the caller thread. Neither is a cancellation. An interrupt does not cancel a Spark job, `DAGScheduler.runJob` waits on the `JobWaiter` until it is cancelled explicitly, so with `useActiveSparkSession=true` a batch pipeline was never cancelled and `cancel()` still reported CANCELLED. `SparkSessionFactory.getOrCreateSession` used `Builder.getOrCreate`, which adopts a usable default session, so a pipeline could stop a session shared with other pipelines in the JVM. Batch `EvaluationContext.stop()` was a no-op. The execution thread kept translating on a stopped SparkContext, which is how the first Spark Versions run of #40090 failed. **Fix, shared code compiled for Spark 3 and Spark 4** - `SparkSessionFactory.acquire` records whether the runner created the session (no usable active or default session before `getOrCreate`) and counts users per created session. `release` stops a created session when its last user leaves. `useActiveSparkSession=true` never owns. Spark 3.5 returns stopped sessions from `getActiveSession` and `getDefaultSession`, so the usable check filters them the way `getOrCreate` does. - The execution thread sets a job group before translation, skips `evaluate()` when cancel arrived during translation, and releases the session in `finally`. The session stop therefore always happens after the last Spark call of the pipeline, on the thread that made it. - Batch `EvaluationContext.stop()` sets a flag the leaf loop checks. Running jobs are cancelled through `cancelJobGroup` with `interruptOnCancel`. `StreamingEvaluationContext` keeps stopping its queries. - `cancel()` stops the context, cancels the job group, joins the execution thread without a timeout, and reports CANCELLED. The join is bounded by Spark itself, task interruption for batch and `spark.sql.streaming.stopTimeout` for queries. A pipeline that already ended reports DONE or FAILED. No `Future.cancel(true)`, no terminal state callback. `waitUntilFinish` reports CANCELLED for a pipeline that ended after a cancel request. **Tests, no sleeps** - `SparkStructuredStreamingPipelineResultTest`: the join returns only after the execution ended, a second `cancel()` is a no-op, a cancelled execution that ends with an exception is CANCELLED, not FAILED. - `StructuredStreamingPipelineStateTest`: a batch cancel of a running job stops the job and the owned session; a pipeline on a session it did not create leaves that session running after cancel. **Behavior changes to note, Spark 3 and Spark 4 structured streaming** - `cancel()` blocks until the execution thread has ended. Tasks that ignore interruption delay it until Spark ends them, `spark.task.reaper.enabled` bounds that. Translation is not interrupted, a cancel during translation skips the evaluation. - `waitUntilFinish()` after a cancel returns CANCELLED instead of throwing. - A session is stopped only when created by the runner, on the execution thread, after its last pipeline. - The Spark UI shows the pipeline's jobs under the group `Beam <jobName>`. - `SparkSessionFactory.getOrCreateSession` is replaced by `acquire` and `release`. The class is not annotated `@Internal`, the only caller in the repository was the runner. Gates: ErrorProne on native JDK 17, checkstyle, spotbugs, spotless, `:runners:spark:4:test` full, `:runners:spark:3:test` for the touched classes, and `StructuredStreamingPipelineStateTest` ten times in the CI fork mode. R: @Abacn -- 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]
