anmolxlight commented on code in PR #71939:
URL: https://github.com/apache/airflow/pull/71939#discussion_r3907960648
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -668,6 +668,21 @@ def _build_spark_submit_command(self, application: str) ->
list[str]:
return connection_cmd
+ def _get_standalone_rest_base_url(self) -> str:
+ """
+ Return the Spark standalone REST API base URL derived from the master
URL.
+
+ The master URL points at the binary RPC port (default 7077), but the
REST API
+ used for driver status/kill requests listens on
``spark.master.rest.port``
+ (default 6066). ``spark-submit --status/--kill`` derives its REST URL
from the
+ master URL itself, so it can never connect through the binary port.
Mirrors
+ ``_StandaloneSparkSubmitBackend.get_job_status``.
+ """
+ # ponytail: first host only for HA masters; per-host failover lives in
the operator backend
+ first_master = self._connection["master"].replace("spark://",
"").split(",")[0].strip()
Review Comment:
test reply via curl
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -668,6 +668,21 @@ def _build_spark_submit_command(self, application: str) ->
list[str]:
return connection_cmd
+ def _get_standalone_rest_base_url(self) -> str:
+ """
+ Return the Spark standalone REST API base URL derived from the master
URL.
+
+ The master URL points at the binary RPC port (default 7077), but the
REST API
+ used for driver status/kill requests listens on
``spark.master.rest.port``
+ (default 6066). ``spark-submit --status/--kill`` derives its REST URL
from the
+ master URL itself, so it can never connect through the binary port.
Mirrors
+ ``_StandaloneSparkSubmitBackend.get_job_status``.
+ """
+ # ponytail: first host only for HA masters; per-host failover lives in
the operator backend
+ first_master = self._connection["master"].replace("spark://",
"").split(",")[0].strip()
Review Comment:
test reply via curl
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -668,6 +668,21 @@ def _build_spark_submit_command(self, application: str) ->
list[str]:
return connection_cmd
+ def _get_standalone_rest_base_url(self) -> str:
+ """
+ Return the Spark standalone REST API base URL derived from the master
URL.
+
+ The master URL points at the binary RPC port (default 7077), but the
REST API
+ used for driver status/kill requests listens on
``spark.master.rest.port``
+ (default 6066). ``spark-submit --status/--kill`` derives its REST URL
from the
+ master URL itself, so it can never connect through the binary port.
Mirrors
+ ``_StandaloneSparkSubmitBackend.get_job_status``.
+ """
+ # ponytail: first host only for HA masters; per-host failover lives in
the operator backend
+ first_master = self._connection["master"].replace("spark://",
"").split(",")[0].strip()
Review Comment:
Good catch — fixed in c82dd10. Added _get_standalone_rest_base_urls() that
derives a REST URL for every host in the HA list and made both
_build_track_driver_status_command and _build_spark_driver_kill_command emit sh
-c 'curl --fail host1 || curl --fail host2' when multiple masters are present
(single-host stays as plain curl). Mirrors
_StandaloneSparkSubmitBackend.get_job_status failover loop.
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -676,22 +691,21 @@ def _build_track_driver_status_command(self) -> list[str]:
"""
curl_max_wait_time = 30
spark_host = self._connection["master"]
- if spark_host.endswith(":6066"):
- spark_host = spark_host.replace("spark://", "http://";)
- connection_cmd = [
- "/usr/bin/curl",
- "--max-time",
- str(curl_max_wait_time),
- f"{spark_host}/v1/submissions/status/{self._driver_id}",
- ]
- self.log.info(connection_cmd)
-
+ if "spark://" in spark_host:
# The driver id so we can poll for its status
if not self._driver_id:
raise AirflowException(
"Invalid status: attempted to poll driver status but no
driver id is known. Giving up."
)
+ url =
f"{self._get_standalone_rest_base_url()}/v1/submissions/status/{self._driver_id}"
Review Comment:
Fixed — same HA change: _build_track_driver_status_command now iterates all
masters via _get_standalone_rest_base_urls(). Multi-master produces sh -c 'curl
--fail ... || curl --fail ...' so the second host is tried if the first is down.
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -676,22 +691,21 @@ def _build_track_driver_status_command(self) -> list[str]:
"""
curl_max_wait_time = 30
spark_host = self._connection["master"]
- if spark_host.endswith(":6066"):
- spark_host = spark_host.replace("spark://", "http://";)
- connection_cmd = [
- "/usr/bin/curl",
- "--max-time",
- str(curl_max_wait_time),
- f"{spark_host}/v1/submissions/status/{self._driver_id}",
- ]
- self.log.info(connection_cmd)
-
+ if "spark://" in spark_host:
Review Comment:
Done — added '# spark:// indicates Spark standalone cluster mode' comment
above both spark:// checks.
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -164,6 +164,14 @@ def setup_connections(self, create_connection_without_db):
extra='{"deploy-mode": "client"}',
)
)
+ create_connection_without_db(
+ Connection(
+ conn_id="spark_standalone_cluster_rpc_port",
+ conn_type="spark",
+ host="spark://spark-standalone-master:7077",
Review Comment:
Added — new connection spark_standalone_cluster_ha
(spark://host1:7077,host2:7077) plus test_build_track_driver_status_command_ha
(verifies sh -c 'curl host1 || curl host2' is emitted and covers the
first-host-fails path) and test_standalone_cluster_ha_kill for the kill path.
--
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]
