You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "hunnguye (via GitHub)" <gi...@apache.org> on 2026/08/21 14:37:41 UTC
[PR] Fix/spark submit standalone status polling #71939 [airflow]
hunnguye opened a new pull request, #71949:
URL: https://github.com/apache/airflow/pull/71949
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
* closes #71881
# Short Description
Using the SparkSubmitOperator to submit an applicaiton to a Spark standalone cluster results in `Failed to poll for the driver status 10 times: returncode = 1` even though the job finished correctly on the spark Server side.
The issue is caused by the current driver-status polling logic, which derives its behavior from the configured master URL but does not sufficiently distinguish between a master URL pointing to the RPC endpoint on port 7077 and one pointing to the REST endpoint on port 6066.
# Changes:
* add a rest_endpoint attribute into the connection dictionary, this contains the correct rest endpoint
* modified and extend _build_spark_driver_kill_command and _build_track_driver_status_command to apply the corrected rest_endpoint logic with minimal change to current behaviour
* new rest_endpoint information has also been applied to the the operator job status request
# Tests:
* extended test cases with separate connection id for the rpc endpoint at 7077
* fully tested against live spark standalone cluster successfully (status polling, killing, job retry)
# Notes
* [operators/spark_submit.py](https://github.com/apache/airflow/compare/main...hunnguye:airflow:fix/spark-submit-standalone-status-polling?expand=1#diff-e814d66fc380bd97c78bab6653452aebadf73d8b7ac7cadec52138a785d30664) assumes a comma separated list of master urls, but as I understand that is not possible as the hook itself always assumes a single url. I therefore ignored that assumption.
* Although the kill operation could be handled entirely through the REST endpoint, I preserved the existing binary kill command to avoid unintended side effects.
---
##### Was generative AI tooling used to co-author this PR?
- [X ] Yes (ChatGPT for brainstorming. I wrote the code most of the code, ChatGPT refactored minimally)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3887749823
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,16 @@ 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"
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
As I understand, the master endpoint is taken from connection.host
https://github.com/apache/airflow/blob/166a1ba1eecf969ed33ad17dd1232b20d7ce5aea/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py#L419-L424
And the whole hook is assuming it is a single host entry:
for example:
https://github.com/apache/airflow/blob/166a1ba1eecf969ed33ad17dd1232b20d7ce5aea/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py#L427
https://github.com/apache/airflow/blob/166a1ba1eecf969ed33ad17dd1232b20d7ce5aea/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py#L561
And the og build_track_driver_command as well (which we will fix in this PR)
https://github.com/apache/airflow/blob/166a1ba1eecf969ed33ad17dd1232b20d7ce5aea/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py#L679
I could fix my code to handle single and multi master endpoints as you requested, but this feels inconsistent with the rest of the current code and could confuse other contributors.
I'd rather suggest to add a comment, and change this in a later PR to handle comma separated master endpoints (and therefore rest endpoints).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on PR #71949:
URL: https://github.com/apache/airflow/pull/71949#issuecomment-5400463821
Thanks for your supportive insights and review!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3849336007
##########
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:
I understand the compatibility concern. However, branching on literal ports is not safer: custom master ports leave `rest_endpoint` unset, while the current `get_job_status()` already supports custom `rest_scheme` and `rest_port`. This change also leaves several existing `get_job_status()` tests failing.
I would prefer deriving the REST endpoint(s) from the master host(s) plus those existing settings, without `6066`/`7077` conditionals, while preserving the current `get_job_status()` behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3887759713
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -1296,17 +1306,30 @@ 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
+ if self._connection["rest_endpoint"]:
+ 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}",
+ ]
+ self.log.info(connection_cmd)
Review Comment:
should i also change the original logging of the status polling to debug ?
@
https://github.com/apache/airflow/blob/166a1ba1eecf969ed33ad17dd1232b20d7ce5aea/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py#L687
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3854012588
##########
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:
I approached this issue, by setting the rest_endpoint first if possible.
Then in status polling and the on_kill process, i just check whether there is a rest_endpoint entry to either run the curl command against that endpoint or alternatively fall back to the binary spark submit
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r4157255317
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
Review Comment:
I think this module has historically grown to only handle single rest_endpoints. I think, the remaining code for the hook only considers a single master endpoint and if we want to consistently handle HA, that would need a bigger refactor (see https://github.com/apache/airflow/pull/71949#discussion_r3887749823).
Although, I can agree with both you and aaron-y-chen, if I'd introduce handling HA in this particular place, the hook would feel inconsistant and maybe confusing to the next contributer.
Regarding your other question:
Custom rest_endpoint altogether as in the master sits at master:7077 but the rest endpoint sits at rest:6066 or even rest-endpoint (without any specific port)? Funny you mentioned this, because that is also one of the original issue we discovered and worked around, until we discovered the polling issue and decided to open a PR to fix that at least.
I think this is a bit of a dilemma, because the contributors before me already suggested rest_scheme and rest_port, which somewhat assumes that the REST endpoint is derived from the master host.
I agree that a fully custom rest_endpoint would be more flexible though. Jowever I'd prefer to keep this PR focused on the polling issue i originally diagnosed and set out to fix. If there is a concrete need for fully customizable endpoints, I think this could be added in a future PR.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on PR #71949:
URL: https://github.com/apache/airflow/pull/71949#issuecomment-5371614709
I did notice PR #71939. However, by that point I had already finished the implementation on my side and only had testing left.
I believe both approaches are very similar. My implementation additionally reuses the REST endpoint logic in operator/spark_submit.py to avoid duplicating the endpoint construction.
One notable difference I noticed is that my implementation uses a POST request to kill the driver, following Spark's [REST API behavior](https://spark.apache.org/docs/latest/spark-standalone.html). PR #71939 appears to use DELETE. I'm not sure whether DELETE is also supported here, but POST is what I tested successfully against the Spark Standalone REST endpoint.
Besides that there are some minor implementation differences.
I’d be happy to compare the approaches and collaborate on a combined solution.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3914169393
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -1296,17 +1306,30 @@ 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
+ if self._connection["rest_endpoint"]:
+ 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}",
+ ]
+ self.log.info(connection_cmd)
Review Comment:
Yes, I think so.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71939 [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #71949:
URL: https://github.com/apache/airflow/pull/71949#issuecomment-5371334927
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our [Contributors' Guide](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example Dag that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "Nataneljpwd (via GitHub)" <gi...@apache.org>.
Nataneljpwd commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r4148990327
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
Review Comment:
can you implement it to use HA? should not be too hard, should just be a small change where the curl is retried, maybe in a loop or a helper command
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -320,41 +352,81 @@ def test_resolve_spark_submit_env_vars_use_krb5ccache_missing_KRB5CCNAME_env(sel
):
hook._build_spark_submit_command(self._spark_job_file)
- def test_build_track_driver_status_command(self):
+ @pytest.mark.parametrize(
+ ("conn_id", "expected_command"),
+ [
+ (
+ "spark_standalone_cluster",
+ [
+ "/usr/bin/curl",
+ "--max-time",
+ "30",
+ "http://spark-standalone-master:6066/v1/submissions/status/driver-1",
+ ],
+ ),
+ (
+ "spark_yarn_cluster",
+ [
+ "spark-submit",
+ "--master",
+ "yarn://yarn-master",
+ "--status",
+ "driver-1",
+ ],
+ ),
+ (
+ "spark_standalone_cluster_rpc_endpoint",
+ [
+ "/usr/bin/curl",
+ "--max-time",
+ "30",
+ "http://spark-standalone-master-rpc-endpoint:6066/v1/submissions/status/driver-1",
+ ],
+ ),
+ (
+ "spark_standalone_cluster_ha",
+ [
+ "spark-submit",
+ "--master",
+ "spark://m1:6066,m2:6066",
+ "--status",
+ "driver-1",
+ ],
+ ),
+ (
+ "spark_standalone_cluster_ipv6",
+ [
+ "/usr/bin/curl",
+ "--max-time",
+ "30",
+ "http://[2001:db8::1]:6066/v1/submissions/status/driver-1",
+ ],
+ ),
+ (
+ "spark_standalone_cluster_ipv6_ha",
+ [
+ "spark-submit",
+ "--master",
+ "spark://[2001:db8::1]:6066,[1993:db8::1]:6066",
+ "--status",
+ "driver-1",
+ ],
+ ),
+ ],
+ )
+ def test_build_track_driver_status_command(self, conn_id, expected_command):
# note this function is only relevant for spark setup matching below condition
# 'spark://' in self._connection['master'] and self._connection['deploy_mode'] == 'cluster'
# Given
- hook_spark_standalone_cluster = SparkSubmitHook(conn_id="spark_standalone_cluster")
- 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 = SparkSubmitHook(conn_id=conn_id)
+ hook._driver_id = "driver-1"
Review Comment:
I like this change done to the test, looks great and way better, thank you!
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -487,6 +560,7 @@ def test_resolve_connection_yarn_default_connection(self):
"keytab": None,
"rest_scheme": "http",
"rest_port": 6066,
+ "rest_endpoint": None,
Review Comment:
some of the connections you changed look very similar, maybe it is worth creating some kind of fixture for it?
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -747,10 +829,63 @@ def test_resolve_connection_spark_standalone_cluster_connection(self):
"keytab": None,
"rest_scheme": "http",
"rest_port": 6066,
+ "rest_endpoint": "http://spark-standalone-master:6066",
}
assert connection == expected_spark_connection
assert cmd[0] == "spark-submit"
+ @pytest.mark.parametrize(
+ ("master", "rest_scheme", "rest_port", "expected"),
+ [
+ (
+ "spark://spark-standalone-master-rpc-endpoint:7078",
+ "http",
+ 6067,
+ "http://spark-standalone-master-rpc-endpoint:6067",
+ ),
+ (
+ "spark://spark-standalone-master-rpc-endpoint:7077",
+ "https",
+ 7443,
+ "https://spark-standalone-master-rpc-endpoint:7443",
+ ),
+ ],
+ )
+ def test_resolve_connection_spark_standalone_cluster_connection_rpc_endpoint(
+ self,
+ create_connection_without_db,
+ master,
+ rest_scheme,
+ rest_port,
+ expected,
+ ):
+ create_connection_without_db(
+ Connection(
+ conn_id="spark_standalone_cluster_rpc_endpoint_parametrized",
+ conn_type="spark",
+ host=master,
+ extra={
+ "deploy-mode": "cluster",
+ "rest-scheme": rest_scheme,
+ "rest-port": rest_port,
Review Comment:
would be nice to set the base rest endpoint here as well
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
Review Comment:
also, what if I have a custom rest_endpoint? what if it's not the same as the connection or master? as if some api gateway routing to a different place per submission and status? this is the reason I liked the direction more than the competing PR, yet HA ans setting a custom rest_endpoint are what's required IMO
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3979655611
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
Not sure whether it made it better.
Hope we are all done now though =)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3846905287
##########
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:
Excuse my ignorance, I was wondering, the _connection attribute is taken from the hook which is generated here:
https://github.com/apache/airflow/blob/c081c558f1c1edc9771d9d136995764d24251834/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py#L395
And seeing it, i didn't thought a comma separated list of master hosts would be possible. Or is this for future proofing? If so I will revert my change / extend the logic for the rest_endpoint to include multiple hosts
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3946072668
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
I think we should also handle IPv6 cases, for example, this host would be broken by this PR `spark://[2001:db8::1]:6066`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #71949:
URL: https://github.com/apache/airflow/pull/71949#issuecomment-5656027431
Note to reviewer: https://github.com/apache/airflow/pull/71939 is a competing PR
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3933973605
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,16 @@ 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"
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
similiarly i also added a regression test for the driver status now
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3915659822
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,16 @@ 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"
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
I think the HA regression can be demonstrated with a small unit test:
```python
def test_standalone_cluster_kill_preserves_ha_master_list(self, create_connection_without_db):
create_connection_without_db(
Connection(
conn_id="spark_standalone_cluster_ha",
conn_type="spark",
host="spark://m1:6066,m2:6066",
extra={"deploy-mode": "cluster"},
)
)
hook = SparkSubmitHook(conn_id="spark_standalone_cluster_ha")
hook._driver_id = "driver-123"
assert hook._build_spark_driver_kill_command() == [
"spark-submit",
"--master",
"spark://m1:6066,m2:6066",
"--kill",
"driver-123",
]
```
This passes before this PR because the complete master list is passed to Spark. On the current head it fails because the generated curl command targets only m1.
Please preserve that behavior, either by retaining `spark-submit --kill` for multi-master configurations or by implementing equivalent per-master REST fallback.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3846871037
##########
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:
I wasn’t sure how much I could /should change, so I opted for a minimal-change PR where the changes would only apply to this specific case when the host is targeting the RPC endpoint 7077.
Especially the logic for the 6066 branch I took from the original code snippet found in the _build_track_driver_status_command
(https://github.com/apache/airflow/pull/71949/changes/BASE..7efaf2dca9a88da31e1758622e3a6be83b70b90f#diff-0babb37b3dc9c7b8e967f3c17d012538f06e28f2f8f4cc66d2a8d9a69bfed115L680)
So, should I build the rest_endpoint using rest_scheme and rest_port, and use that endpoint for both status polling and driver kills, effectively replacing the current spark-submit --status / --kill commands?
Since this issue has been open for quite a long time (https://github.com/apache/airflow/issues/20949), I was concerned that introducing broader or potentially breaking changes might affect users who have already found workarounds for the current behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3854019060
##########
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:
As you suggested i fixed the CI tests by adding the rest_endpoint attribute where applicable.
I also added a parameterized regression test for the examples you provided
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal merged PR #71949:
URL: https://github.com/apache/airflow/pull/71949
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r4157260434
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -487,6 +560,7 @@ def test_resolve_connection_yarn_default_connection(self):
"keytab": None,
"rest_scheme": "http",
"rest_port": 6066,
+ "rest_endpoint": None,
Review Comment:
I think i can figure something out :)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r4166139772
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -487,6 +560,7 @@ def test_resolve_connection_yarn_default_connection(self):
"keytab": None,
"rest_scheme": "http",
"rest_port": 6066,
+ "rest_endpoint": None,
Review Comment:
I introduced something like a default set of connection attributes, which I took from the _resolve_connection method.
Each test would then just override the necessary parameters.
Tests completed without failure.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3853990276
##########
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:
I reverted my changes to the get_job_status method
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3885487329
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,16 @@ 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"
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
If `master="spark://m1:7077,m2:7077"`, then `rest_endpoint` will become `"http://m1:6066"`, which doesn't seem to work as expected.
Please fix it and add a regression test for this case.
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -1296,17 +1306,30 @@ 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
+ if self._connection["rest_endpoint"]:
+ 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}",
+ ]
+ self.log.info(connection_cmd)
Review Comment:
nit: maybe we should use debug level log here.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3975003484
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
Maybe parameterize some of the tests?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #71949:
URL: https://github.com/apache/airflow/pull/71949#issuecomment-6063620706
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "Nataneljpwd (via GitHub)" <gi...@apache.org>.
Nataneljpwd commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r4220815513
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
Review Comment:
you may resolve
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "aaron-y-chen (via GitHub)" <gi...@apache.org>.
aaron-y-chen commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3849204937
##########
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:
Yes, `master` can contain multiple comma-separated master endpoints in Spark Standalone HA ([ref](https://spark.apache.org/docs/3.5.6/spark-standalone.html#standby-masters-with-zookeeper)). The existing HA unit tests cover failover across these endpoints, and all three currently fail after this change. Please preserve the per-master failover behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on PR #71949:
URL: https://github.com/apache/airflow/pull/71949#issuecomment-5412132840
Although `_StandaloneSparkSubmitBackend.get_job_status()` supports multiple master URLs in connection["master"], I chose to follow the hook’s current assumption that connection["master"] contains a single URL and constructed connection["rest_endpoint"] accordingly.
I could also have added support for rest_endpoint to handle either a single URL or a comma-separated list of master urls, but that would conflict with the assumption above. I think doing this properly would also require refactoring the usages of connection["master"], either always assume a comma separated list of master url of datatype string or already providing a list of master urls (which results in just one element if not in HA). However, that feels outside the scope of this PR.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix/spark submit standalone status polling #71881 [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3915892050
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,16 @@ 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"
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
Ok no problem.
In that case i would, explicitely only create the rest_endpoint attribute, if there is just one elment in a comma separated list. That way we can pass the test, preserve and stay consistent with the current assumption of the master endpoint while still leave room for later adjustments and refactoring for multi master endpoints
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "hunnguye (via GitHub)" <gi...@apache.org>.
hunnguye commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r3967697620
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
+ ):
+ host = conn_data["master"].replace("spark://", "").strip()
+ conn_data["rest_endpoint"] = (
+ f"{conn_data['rest_scheme']}://{host.split(':')[0]}:{conn_data['rest_port']}"
Review Comment:
nice catch.
I added some other tests for this case of handling ipv6 endpoints.
To be frank, i feel like the tests are a bit crowded now or could be optimized, but im not familiar with all the possible tests nor would that fall into the scope of this PR.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Spark submit standalone status polling to use correct REST endpoint [airflow]
Posted by "Nataneljpwd (via GitHub)" <gi...@apache.org>.
Nataneljpwd commented on code in PR #71949:
URL: https://github.com/apache/airflow/pull/71949#discussion_r4220811872
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -457,6 +458,17 @@ 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"
+ and "," not in conn_data["master"] # only consider single master, non-HA for now
Review Comment:
sure, no problem, I agree, we can keep it non HA for now, and in a future pr, make it HA, and possibly allow for fully customizable endpoints in another future PR
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org