This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 239b8a2c8dcd [SPARK-57613][PYTHON] Stop the Spark session when a
Declarative Pipelines run step fails
239b8a2c8dcd is described below
commit 239b8a2c8dcd8821184c999af8389fb3d3aa4c75
Author: YangJie <[email protected]>
AuthorDate: Mon Jun 29 11:38:50 2026 +0800
[SPARK-57613][PYTHON] Stop the Spark session when a Declarative Pipelines
run step fails
### What changes were proposed in this pull request?
The Declarative Pipelines CLI `run()` (in
`python/pyspark/pipelines/cli.py`) creates a Spark Connect session and
previously stopped it only in a `finally` that wrapped
`handle_pipeline_events`. Graph creation, element registration, and starting
the run all happened before that `try`, so if any of them failed the session
was never stopped and leaked. This moves those steps inside the `try` so the
`finally: spark.stop()` runs once the session exists, regardless of where a
later step fails.
### Why are the changes needed?
A failure in `create_dataflow_graph`, `register_definitions`, or
`start_run` left the Spark Connect session running. For a short-lived CLI
invocation that means a leaked session (and its server-side resources) on every
failed run.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass Github Actions
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56668 from LuciferYang/sdp-cli-session-leak-on-failure.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
python/pyspark/pipelines/cli.py | 47 +++++++++++++++++++++--------------------
1 file changed, 24 insertions(+), 23 deletions(-)
diff --git a/python/pyspark/pipelines/cli.py b/python/pyspark/pipelines/cli.py
index bc0d718f8a75..d98e189611bd 100644
--- a/python/pyspark/pipelines/cli.py
+++ b/python/pyspark/pipelines/cli.py
@@ -325,30 +325,31 @@ def run(
spark_builder = spark_builder.config(key, value)
spark = spark_builder.getOrCreate()
-
- log_with_curr_timestamp("Creating dataflow graph...")
- dataflow_graph_id = create_dataflow_graph(
- spark,
- default_catalog=spec.catalog,
- default_database=spec.database,
- sql_conf=spec.configuration,
- )
-
- log_with_curr_timestamp("Registering graph elements...")
- registry = SparkConnectGraphElementRegistry(spark, dataflow_graph_id)
- register_definitions(spec_path, registry, spec, spark, dataflow_graph_id)
-
- log_with_curr_timestamp("Starting run...")
- result_iter = start_run(
- spark,
- dataflow_graph_id,
- full_refresh=full_refresh,
- full_refresh_all=full_refresh_all,
- refresh=refresh,
- dry=dry,
- storage=spec.storage,
- )
+ # Stop the session even if graph creation, registration, or the run itself
fails, so a failure
+ # after the session is created does not leak it.
try:
+ log_with_curr_timestamp("Creating dataflow graph...")
+ dataflow_graph_id = create_dataflow_graph(
+ spark,
+ default_catalog=spec.catalog,
+ default_database=spec.database,
+ sql_conf=spec.configuration,
+ )
+
+ log_with_curr_timestamp("Registering graph elements...")
+ registry = SparkConnectGraphElementRegistry(spark, dataflow_graph_id)
+ register_definitions(spec_path, registry, spec, spark,
dataflow_graph_id)
+
+ log_with_curr_timestamp("Starting run...")
+ result_iter = start_run(
+ spark,
+ dataflow_graph_id,
+ full_refresh=full_refresh,
+ full_refresh_all=full_refresh_all,
+ refresh=refresh,
+ dry=dry,
+ storage=spec.storage,
+ )
handle_pipeline_events(result_iter)
finally:
spark.stop()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
