aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3843739745
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -1296,17 +1310,31 @@ def _build_spark_driver_kill_command(self) -> list[str]:
:return: full command to kill a driver
"""
- # Assume that spark-submit is present in the path to the executing user
- connection_cmd = [self._connection["spark_binary"]]
+ curl_max_wait_time = 30
+ spark_host = self._connection["master"]
+ if spark_host.endswith(":7077"):
+ spark_host = self._connection["rest_endpoint"]
+ connection_cmd = [
+ "/usr/bin/curl",
+ "--max-time",
+ str(curl_max_wait_time),
+ "-X",
+ "POST",
+ f"{spark_host}/v1/submissions/kill/{self._driver_id}",
+ ]
Review Comment:
This fixes driver termination only when the RPC port is exactly `7077`. A
standalone master using a custom RPC port still takes the broken `spark-submit
--kill` path.
Once standalone mode is known, always send `POST
/v1/submissions/kill/{driver_id}` to the configured REST endpoint,
independently of the master port.
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,19 @@ def _resolve_connection(self) -> dict[str, Any]:
conn_data["keytab"] =
self._create_keytab_path_from_base64_keytab(
base64_keytab, conn_data["principal"]
)
+ # Construct the Standalone Restendpoint
+ if (
+ conn.conn_type == "spark"
+ and conn_data["master"].startswith("spark://")
+ and conn_data["deploy_mode"] == "cluster"
+ ):
+ if conn_data["master"].endswith("7077"):
Review Comment:
Please do not infer the endpoint type from literal ports. Both the RPC and
REST ports are configurable, so `spark://host:7078` leaves `rest_endpoint`
unset and reproduces the original failure. The `6066` branch also ignores
`rest_scheme=https`.
Please derive the REST URL from the master host plus the existing connection
settings for every standalone master:
```python
base_url = f"{rest_scheme}://{master_host}:{rest_port}"
```
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -329,6 +337,10 @@ def test_build_track_driver_status_command(self):
hook_spark_standalone_cluster._driver_id = "driver-20171128111416-0001"
hook_spark_yarn_cluster = SparkSubmitHook(conn_id="spark_yarn_cluster")
hook_spark_yarn_cluster._driver_id = "driver-20171128111417-0001"
+ hook_spark_standalone_cluster_rpc_endpoint = SparkSubmitHook(
Review Comment:
Please add parametrized regression coverage for custom RPC/REST ports,
HTTPS, and HA fallback. The current `7077 -> 6066` case only covers the
defaults and would not catch the regressions above.
For example:
```python
@pytest.mark.parametrize(
("master", "rest_scheme", "rest_port", "expected"),
[
("spark://host:7078", "http", 6067, "http://host:6067";),
("spark://host:7077", "https", 7443, "https://host:7443";),
],
)
```
##########
providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:
##########
@@ -212,22 +212,16 @@ def submit_job(self, context: Context) -> str | None:
return driver_id
def get_job_status(self, external_id: str, context: Context) -> str:
- scheme = self.hook._connection.get("rest_scheme", "http")
- rest_port = self.hook._connection.get("rest_port", 6066)
- # HA master URLs can look like spark://m1:7077,m2:7077 — try each host
in order.
- # The master URL port (e.g. 7077) is the RPC port — not the REST API
port.
- # Use rest-port connection extra to override spark.master.rest.port
(default 6066).
- master_urls = self.hook._connection["master"].replace("spark://",
"").split(",")
+
last_exc: Exception = RuntimeError("No Spark masters to query")
- for m in master_urls:
Review Comment:
This loop provides failover across Standalone masters and should be
preserved. Replacing it with a single `rest_endpoint` breaks HA when that
endpoint is unavailable.
--
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]
