From 77a86111f0e9de58a29bfc3d06c373582dbe7d96 Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Tue, 21 Jul 2026 16:01:13 +0530 Subject: [PATCH 1/2] Include spark submit canonical logs in failure exceptions --- .../apache/spark/hooks/spark_submit.py | 19 ++++++ .../apache/spark/hooks/test_spark_submit.py | 59 +++++++++++++++++++ 2 files changed, 78 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 cd4a0703d19da..12f1b9ddbb2ed 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 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 @@ def __init__( 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,17 @@ def _mask_cmd(self, connection_cmd: str | list[str]) -> str: return connection_cmd_masked + def _format_submit_log_tail(self) -> str: + """ + Capture 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 +796,11 @@ def submit(self, application: str = "", **kwargs: Any) -> str | None: 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._format_submit_log_tail()}" ) raise AirflowException( f"Cannot execute: {self._mask_cmd(spark_submit_cmd)}. Error code is: {returncode}." + f"{self._format_submit_log_tail()}" ) if self._should_track_yarn_application_via_rm_api(): @@ -794,6 +811,7 @@ def submit(self, application: str = "", **kwargs: Any) -> str | None: 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._format_submit_log_tail()}" ) finally: # K8s-API tracking defers post-submit commands to _poll_k8s_driver_via_api's finally @@ -866,6 +884,7 @@ def _process_spark_submit_log(self, itr: Iterator[Any]) -> None: 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 5733f44fd6b27..7bb03701f6767 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 @@ def test_spark_process_runcmd(self, mock_popen, sdk_connection_not_found): 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 @@ def test_process_spark_submit_log_standalone_cluster(self): 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,20 @@ def test_masks_passwords(self, command: str, expected: str) -> None: # Then assert command_masked == expected + @pytest.mark.db_test + def test_format_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._format_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() From 94feeface64980147dcdb36b5a4c87042e0c742a Mon Sep 17 00:00:00 2001 From: Amogh Desai Date: Wed, 22 Jul 2026 17:05:34 +0530 Subject: [PATCH 2/2] comments from wei --- .../providers/apache/spark/hooks/spark_submit.py | 11 ++++++----- .../unit/apache/spark/hooks/test_spark_submit.py | 10 ++++++++-- 2 files changed, 14 insertions(+), 7 deletions(-) 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 12f1b9ddbb2ed..5e198044aa564 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 @@ -534,9 +534,10 @@ def _mask_cmd(self, connection_cmd: str | list[str]) -> str: return connection_cmd_masked - def _format_submit_log_tail(self) -> str: + @property + def _submit_log_tail(self) -> str: """ - Capture the last few lines of the spark-submit process's own output. + 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. """ @@ -796,11 +797,11 @@ def submit(self, application: str = "", **kwargs: Any) -> str | None: 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._format_submit_log_tail()}" + f"{self._submit_log_tail}" ) raise AirflowException( f"Cannot execute: {self._mask_cmd(spark_submit_cmd)}. Error code is: {returncode}." - f"{self._format_submit_log_tail()}" + f"{self._submit_log_tail}" ) if self._should_track_yarn_application_via_rm_api(): @@ -811,7 +812,7 @@ def submit(self, application: str = "", **kwargs: Any) -> str | None: 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._format_submit_log_tail()}" + f"{self._submit_log_tail}" ) finally: # K8s-API tracking defers post-submit commands to _poll_k8s_driver_via_api's finally 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 7bb03701f6767..45555a1186fcf 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 @@ -1286,12 +1286,18 @@ def test_masks_passwords(self, command: str, expected: str) -> None: assert command_masked == expected @pytest.mark.db_test - def test_format_submit_log_tail_formats_and_masks_captured_lines(self) -> None: + 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._format_submit_log_tail() + tail = hook._submit_log_tail assert tail == ( "\nLast spark-submit output:\n"