shahar1 commented on code in PR #70513:
URL: https://github.com/apache/airflow/pull/70513#discussion_r3656532753
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -885,7 +889,11 @@ 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)
+ if self._exception_anchor_seen:
+ self._last_submit_log_lines.append(line)
+ elif _EXCEPTION_START_RE.search(line):
+ self._exception_anchor_seen = True
+ self._last_submit_log_lines.append(line)
Review Comment:
**major** — when `spark-submit` fails *without* an uncaught JVM exception,
this now captures nothing at all, so `_submit_log_tail` is `""` and the failure
message is back to a bare `Error code is: 1`.
Concrete cases that hit this:
- a failing **PySpark** app — `SparkSubmit.doSubmit` catches
`SparkUserAppException` and calls `exitFn(e.exitCode)`; the Python traceback is
printed but there is no `Exception in thread "..."` line;
- `Error: Failed to load class ...`, `Error: Missing application resource`,
`Error: Master must either be yarn or start with spark, mesos, k8s...` — all go
through `printErrorAndExit`, no marker.
Those all used to surface the real error in the 20-line tail. The two
behaviours don't have to be mutually exclusive — let the anchor *reset* the
buffer rather than gate it:
```suggestion
if not self._exception_anchor_seen and
_EXCEPTION_START_RE.search(line):
# Drop the pre-exception banner noise, keep the whole trace
from here on.
self._exception_anchor_seen = True
self._last_submit_log_lines = deque(maxlen=500)
self._last_submit_log_lines.append(line)
```
This needs `__init__` to go back to `deque(maxlen=20)` (see the comment
there): full trace when there is an exception, last-20 fallback when there
isn't.
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -55,6 +55,11 @@
_K8S_WAIT_APP_COMPLETION_CONF = "spark.kubernetes.submission.waitAppCompletion"
+# JVM's default uncaught exception handler always prints this exact shape,
regardless of
+# which library raised the exception and that makes it a reliable anchor for
where the real error starts,
+# unlike guessing at stack frame formatting.
+_EXCEPTION_START_RE = re.compile(r'Exception in thread "[^"]*"')
Review Comment:
**nit** — three lines of prose for a one-line regex. The *why* (the JVM
prints this exact shape regardless of which library threw) earns its keep;
`unlike guessing at stack frame formatting` is padding. One line would do.
##########
providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py:
##########
@@ -319,9 +324,8 @@ 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._last_submit_log_lines: deque[str] = deque(maxlen=500)
Review Comment:
**minor** — `500` is both unexplained and untested now.
`test_process_spark_submit_log_last_submit_log_lines_truncates_to_maxlen` was
removed and its replacement only exercises 30 frames, so nothing covers the cap.
[`code-review.instructions.md` § Testing
Requirements](https://github.com/apache/airflow/blob/main/.github/instructions/code-review.instructions.md#testing-requirements):
> **Flag any changed or added behaviour without a corresponding test, and
flag tests that a reviewer cannot fail by reverting the PR's change.**
The removed comment also took the buffer's rationale with it — worth a short
note on why 500 (and, per the other comment, this line probably wants to stay
`maxlen=20` with the anchor swapping in the larger buffer).
##########
providers/apache/spark/tests/unit/apache/spark/hooks/test_spark_submit.py:
##########
@@ -391,19 +391,22 @@ def test_submit_failure_includes_captured_log_tail(self,
mock_popen, sdk_connect
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)
+ assert 'Exception in thread "main" org.apache.spark.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):
+ def test_submit_no_driver_id_omits_log_tail_without_exception_marker(
+ self, mock_popen, sdk_connection_not_found
+ ):
+ """spark-submit output with no uncaught-exception marker contributes
no log tail."""
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)
+ assert "Last spark-submit output:" not in str(exc_info.value)
Review Comment:
**minor** — this test was
`test_submit_no_driver_id_includes_captured_log_tail` and asserted the tail
*was* present; it is now inverted to assert its absence.
[`code-review.instructions.md` § Quality Signals to
Check](https://github.com/apache/airflow/blob/main/.github/instructions/code-review.instructions.md#quality-signals-to-check):
> **Flag any existing test modified to accommodate new behavior** — this may
indicate a behavioral regression rather than a genuine fix.
That's exactly the regression flagged on `spark_submit.py:892`. If the
fallback goes in, this test can keep asserting a tail is present, and a new
case can cover the `Error: Failed to load class ...` / PySpark-traceback shape
that carries no `Exception in thread` marker.
--
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]
