This is an automated email from the ASF dual-hosted git repository.
eladkal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 104ad12e190 Include spark submit canonical logs in failure exceptions
(#70178)
104ad12e190 is described below
commit 104ad12e190db7197fa303d98a3ab68879eafd33
Author: Amogh Desai <[email protected]>
AuthorDate: Thu Jul 23 09:02:58 2026 +0530
Include spark submit canonical logs in failure exceptions (#70178)
* Include spark submit canonical logs in failure exceptions
* comments from wei
---
.../providers/apache/spark/hooks/spark_submit.py | 20 +++++++
.../unit/apache/spark/hooks/test_spark_submit.py | 65 ++++++++++++++++++++++
2 files changed, 85 insertions(+)
diff --git
a/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py
b/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py
index cd4a0703d19..5e198044aa5 100644
---
a/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py
+++
b/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py
@@ -27,6 +27,7 @@ import subprocess
import tempfile
import time
import uuid
+from collections import deque
from collections.abc import Iterator
from functools import cached_property
from pathlib import Path
@@ -318,6 +319,9 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
self._driver_id: str | None = None
self._driver_status: str | None = None
self._spark_exit_code: int | None = None
+ # Last few lines of the spark-submit process's own stdout/stderr, so
failure
+ # exceptions can include the actual root cause instead of just an exit
code.
+ self._last_submit_log_lines: deque[str] = deque(maxlen=20)
self._env: dict[str, Any] | None = None
self._post_submit_commands: list[str] = list(post_submit_commands) if
post_submit_commands else []
self._post_submit_commands_done: bool = False
@@ -530,6 +534,18 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
return connection_cmd_masked
+ @property
+ def _submit_log_tail(self) -> str:
+ """
+ The last few lines of the spark-submit process's own output.
+
+ Appended to submit-failure exceptions so the real root cause is
visible instead of just an exit code.
+ """
+ if not self._last_submit_log_lines:
+ return ""
+ tail = "\n".join(self._mask_cmd([line]) for line in
self._last_submit_log_lines)
+ return f"\nLast spark-submit output:\n{tail}"
+
def _build_spark_common_args(self) -> list[str]:
"""
Build common Spark arguments that are shared between spark-submit and
spark-pipelines.
@@ -781,9 +797,11 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
raise AirflowException(
f"Cannot execute: {self._mask_cmd(spark_submit_cmd)}.
Error code is: {returncode}. "
f"Kubernetes spark exit code is:
{self._spark_exit_code}"
+ f"{self._submit_log_tail}"
)
raise AirflowException(
f"Cannot execute: {self._mask_cmd(spark_submit_cmd)}.
Error code is: {returncode}."
+ f"{self._submit_log_tail}"
)
if self._should_track_yarn_application_via_rm_api():
@@ -794,6 +812,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
if self._should_track_driver_status and self._driver_id is None:
raise AirflowException(
"No driver id is known: something went wrong when
executing the spark submit command"
+ f"{self._submit_log_tail}"
)
finally:
# K8s-API tracking defers post-submit commands to
_poll_k8s_driver_via_api's finally
@@ -866,6 +885,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
self._driver_id = match_driver_id.group(0)
self.log.info("identified spark driver id: %s",
self._driver_id)
+ self._last_submit_log_lines.append(line)
self.log.info(line)
def _start_yarn_application_status_tracking(self, application_id: str) ->
None:
diff --git
a/providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py
b/providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py
index 5733f44fd6b..45555a1186f 100644
--- a/providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py
+++ b/providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py
@@ -378,6 +378,33 @@ class TestSparkSubmitHook:
bufsize=-1,
)
+ @pytest.mark.db_test
+
@patch("airflow.providers.apache.spark.hooks.spark_submit.subprocess.Popen")
+ def test_submit_failure_includes_captured_log_tail(self, mock_popen,
sdk_connection_not_found):
+ mock_popen.return_value.stdout = StringIO(
+ "Exception in thread main: SparkException: bad jar\nsome other
line"
+ )
+ mock_popen.return_value.stderr = StringIO("")
+ mock_popen.return_value.wait.return_value = 1
+
+ hook = SparkSubmitHook(conn_id="")
+
+ with pytest.raises(AirflowException, match="Last spark-submit
output:") as exc_info:
+ hook.submit()
+ assert "Exception in thread main: SparkException: bad jar" in
str(exc_info.value)
+
+ @pytest.mark.db_test
+
@patch("airflow.providers.apache.spark.hooks.spark_submit.subprocess.Popen")
+ def test_submit_no_driver_id_includes_captured_log_tail(self, mock_popen,
sdk_connection_not_found):
+ mock_popen.return_value.stdout = StringIO("some unrelated spark-submit
output")
+ mock_popen.return_value.stderr = StringIO("")
+ mock_popen.return_value.wait.return_value = 0
+
+ hook = SparkSubmitHook(conn_id="spark_standalone_cluster")
+ with pytest.raises(AirflowException, match="No driver id is known") as
exc_info:
+ hook.submit()
+ assert "Last spark-submit output:\nsome unrelated spark-submit output"
in str(exc_info.value)
+
@pytest.mark.db_test
def test_resolve_should_track_driver_status(self,
sdk_connection_not_found):
# Given
@@ -986,6 +1013,24 @@ class TestSparkSubmitHook:
assert hook._driver_id == "driver-20171128111415-0001"
+ def test_process_spark_submit_log_populates_last_submit_log_lines(self):
+ hook = SparkSubmitHook(conn_id="spark_standalone_cluster")
+ log_lines = [
+ "Running Spark using the REST application submission protocol.",
+ "17/11/28 11:14:15 INFO RestSubmissionClient: Submitting a request
"
+ "to launch an application in spark://spark-standalone-master:6066",
+ ]
+
+ hook._process_spark_submit_log(log_lines)
+
+ assert list(hook._last_submit_log_lines) == log_lines
+
+ def
test_process_spark_submit_log_last_submit_log_lines_truncates_to_maxlen(self):
+ hook = SparkSubmitHook(conn_id="spark_standalone_cluster")
+ log_lines = [f"line {i}" for i in range(25)]
+ hook._process_spark_submit_log(log_lines)
+ assert list(hook._last_submit_log_lines) == log_lines[-20:]
+
def test_process_spark_driver_status_log(self):
# Given
hook = SparkSubmitHook(conn_id="spark_standalone_cluster")
@@ -1240,6 +1285,26 @@ class TestSparkSubmitHook:
# Then
assert command_masked == expected
+ @pytest.mark.db_test
+ def test_submit_log_tail_empty_when_no_lines_captured(self) -> None:
+ hook = SparkSubmitHook()
+
+ assert hook._submit_log_tail == ""
+
+ @pytest.mark.db_test
+ def test_submit_log_tail_formats_and_masks_captured_lines(self) -> None:
+ hook = SparkSubmitHook()
+ hook._last_submit_log_lines.append("Exception in thread main:
SparkException: bad jar")
+ hook._last_submit_log_lines.append("--password='secret'")
+
+ tail = hook._submit_log_tail
+
+ assert tail == (
+ "\nLast spark-submit output:\n"
+ "Exception in thread main: SparkException: bad jar\n"
+ "--password='******'"
+ )
+
@pytest.mark.db_test
def test_create_keytab_path_from_base64_keytab_with_decode_exception(self):
hook = SparkSubmitHook()
