Include spark submit canonical logs in failure exceptions

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,18 @@ def _mask_cmd(self, connection_cmd: str | list[str]) -> str:
 
         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 @@ 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._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 @@ 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._submit_log_tail}"
                 )
         finally:
             # K8s-API tracking defers post-submit commands to _poll_k8s_driver_via_api's finally
@@ -866,6 +885,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:

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,26 @@ def test_masks_passwords(self, command: str, expected: str) -> None:
         # 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()