You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "amoghrajesh (via GitHub)" <gi...@apache.org> on 2026/07/27 10:17:15 UTC
[PR] Anchor spark-submit log trail on actual exception instead of lines [airflow]
amoghrajesh opened a new pull request, #70513:
URL: https://github.com/apache/airflow/pull/70513
<!-- 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
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
Related to https://github.com/apache/airflow/pull/70178
The previous fix kept a rolling tail of the last 20 lines of spark-submit's own output, but deep JVM stack traces (Fabric8/Vert.x/Netty chains routinely run 70+ lines through nested `Caused by:` blocks) pushed the real exception
message out of the buffer before the process finished, leaving only bare stack frames with no message text. A live K8s quota rejection test reproduced this exactly: the LLM retry policy saw only Netty I/O frames and misclassified
a permanent quota error as a transient SSL issue.
Anchor capture on the JVM's own uncaught-exception marker (`Exception in thread "..."`) instead of a blind line count. Everything before the marker is noise (banners, WARN logs) and is dropped; everything from the marker onward is kept, with a generous safety-valve cap rather than a tight rolling window. If `spark-submit` never throws an uncaught exception,
nothing is captured and the failure message carries no log tail, rather than a plain last-N tail of possibly-irrelevant output.
### Testing
Error thrown now:
```shell
2026-07-27 15:35:25] ERROR - Task failed with exception
AirflowException: Cannot execute: spark-submit --master k8s://https://host.docker.internal:6445 --conf spark.kubernetes.container.image=apache/spark:3.5.3 --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark --conf spark.driver.extraJavaOptions=-Djavax.net.ssl.trustAll=true --conf spark.executor.extraJavaOptions=-Djavax.net.ssl.trustAll=true --conf spark.kubernetes.submission.waitAppCompletion=false --conf spark.kubernetes.namespace=spark --name arrow-spark --class org.apache.spark.examples.SparkPi --deploy-mode cluster local:///opt/spark/examples/jars/spark-examples_2.12-3.5.3.jar 100. Error code is: 1. Kubernetes spark exit code is: None
Last spark-submit output:
Exception in thread "main" io.fabric8.kubernetes.client.KubernetesClientException: Failure executing: POST at: https://host.docker.internal:6445/api/v1/namespaces/spark/pods. Message: pods "arrow-spark-5808b09fa309515b-driver" is forbidden: exceeded quota: spark-demo-quota, requested: count/pods=1, used: count/pods=3, limited: count/pods=3. Received status: Status(apiVersion=v1, code=403, details=StatusDetails(causes=[], group=null, kind=pods, name=arrow-spark-5808b09fa309515b-driver, retryAfterSeconds=null, uid=null, additionalProperties={}), kind=Status, message=pods "arrow-spark-5808b09fa309515b-driver" is forbidden: exceeded quota: spark-demo-quota, requested: count/pods=1, used: count/pods=3, limited: count/pods=3, metadata=ListMeta(_continue=null, remainingItemCount=null, resourceVersion=null, selfLink=null, additionalProperties={}), reason=Forbidden, status=Failure, additionalProperties={}).
at io.fabric8.kubernetes.client.KubernetesClientException.copyAsCause(KubernetesClientException.java:205)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.waitForResult(OperationSupport.java:507)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.handleResponse(OperationSupport.java:524)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.handleCreate(OperationSupport.java:340)
at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.handleCreate(BaseOperation.java:766)
at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.handleCreate(BaseOperation.java:98)
at io.fabric8.kubernetes.client.dsl.internal.CreateOnlyResourceOperation.create(CreateOnlyResourceOperation.java:42)
at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.create(BaseOperation.java:1167)
at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.create(BaseOperation.java:98)
at org.apache.spark.deploy.k8s.submit.Client.run(KubernetesClientApplication.scala:154)
at org.apache.spark.deploy.k8s.submit.KubernetesClientApplication.$anonfun$run$6(KubernetesClientApplication.scala:258)
at org.apache.spark.deploy.k8s.submit.KubernetesClientApplication.$anonfun$run$6$adapted(KubernetesClientApplication.scala:252)
at org.apache.spark.util.SparkErrorUtils.tryWithResource(SparkErrorUtils.scala:51)
at org.apache.spark.util.SparkErrorUtils.tryWithResource$(SparkErrorUtils.scala:48)
at org.apache.spark.util.Utils$.tryWithResource(Utils.scala:97)
at org.apache.spark.deploy.k8s.submit.KubernetesClientApplication.run(KubernetesClientApplication.scala:252)
at org.apache.spark.deploy.k8s.submit.KubernetesClientApplication.start(KubernetesClientApplication.scala:225)
at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:1033)
at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:203)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:226)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:95)
at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1171)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1180)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: io.fabric8.kubernetes.client.KubernetesClientException: Failure executing: POST at: https://host.docker.internal:6445/api/v1/namespaces/spark/pods. Message: pods "arrow-spark-5808b09fa309515b-driver" is forbidden: exceeded quota: spark-demo-quota, requested: count/pods=1, used: count/pods=3, limited: count/pods=3. Received status: Status(apiVersion=v1, code=403, details=StatusDetails(causes=[], group=null, kind=pods, name=arrow-spark-5808b09fa309515b-driver, retryAfterSeconds=null, uid=null, additionalProperties={}), kind=Status, message=pods "arrow-spark-5808b09fa309515b-driver" is forbidden: exceeded quota: spark-demo-quota, requested: count/pods=1, used: count/pods=3, limited: count/pods=3, metadata=ListMeta(_continue=null, remainingItemCount=null, resourceVersion=null, selfLink=null, additionalProperties={}), reason=Forbidden, status=Failure, additionalProperties={}).
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.requestFailure(OperationSupport.java:642)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.requestFailure(OperationSupport.java:622)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.assertResponseCode(OperationSupport.java:582)
at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.lambda$handleResponse$0(OperationSupport.java:549)
at java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:646)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)
at io.fabric8.kubernetes.client.http.StandardHttpClient.lambda$completeOrCancel$10(StandardHttpClient.java:141)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)
at io.fabric8.kubernetes.client.http.ByteArrayBodyHandler.onBodyDone(ByteArrayBodyHandler.java:51)
at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)
at io.fabric8.kubernetes.client.vertx.VertxHttpRequest.lambda$consumeBytes$1(VertxHttpRequest.java:84)
at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:270)
at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:252)
at io.vertx.core.http.impl.HttpEventHandler.handleEnd(HttpEventHandler.java:76)
at io.vertx.core.http.impl.HttpClientResponseImpl.handleEnd(HttpClientResponseImpl.java:250)
at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.lambda$new$0(Http1xClientConnection.java:448)
at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:279)
at io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:157)
at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.handleEnd(Http1xClientConnection.java:758)
at io.vertx.core.impl.ContextImpl.execute(ContextImpl.java:327)
at io.vertx.core.impl.ContextImpl.execute(ContextImpl.java:307)
at io.vertx.core.http.impl.Http1xClientConnection.handleResponseEnd(Http1xClientConnection.java:989)
at io.vertx.core.http.impl.Http1xClientConnection.handleHttpMessage(Http1xClientConnection.java:861)
at io.vertx.core.http.impl.Http1xClientConnection.handleMessage(Http1xClientConnection.java:823)
at io.vertx.core.net.impl.ConnectionBase.read(ConnectionBase.java:159)
at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:153)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:355)
at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:434)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:361)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:325)
at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:249)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:355)
at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1526)
at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1384)
at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1435)
at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:545)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:484)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1429)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:918)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:176)
at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.handle(AbstractNioChannel.java:445)
at io.netty.channel.nio.NioIoHandler$DefaultNioRegistration.handle(NioIoHandler.java:388)
at io.netty.channel.nio.NioIoHandler.processSelectedKey(NioIoHandler.java:596)
at io.netty.channel.nio.NioIoHandler.processSelectedKeysOptimized(NioIoHandler.java:571)
at io.netty.channel.nio.NioIoHandler.processSelectedKeys(NioIoHandler.java:512)
at io.netty.channel.nio.NioIoHandler.run(NioIoHandler.java:484)
at io.netty.channel.SingleThreadIoEventLoop.runIo(SingleThreadIoEventLoop.java:225)
at io.netty.channel.SingleThreadIoEventLoop.run(SingleThreadIoEventLoop.java:196)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:1195)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:840)
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1593 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 195 in wrapper
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 2178 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 195 in wrapper
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 2136 in _run_execute_callable
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 445 in wrapper
File "/opt/airflow/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py", line 292 in execute
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/resumablejobmixin.py", line 200 in execute_resumable
File "/opt/airflow/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py", line 309 in submit_job
File "/opt/airflow/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py", line 801 in submit
▶Post Execute
```
Earlier:
```shell
[2026-07-27 15:14:40] ERROR - Task failed with exception
AirflowException: Cannot execute: spark-submit --master k8s://https://host.docker.internal:6445 --conf spark.kubernetes.container.image=apache/spark:3.5.3 --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark --conf spark.driver.extraJavaOptions=-Djavax.net.ssl.trustAll=true --conf spark.executor.extraJavaOptions=-Djavax.net.ssl.trustAll=true --conf spark.kubernetes.submission.waitAppCompletion=false --conf spark.kubernetes.namespace=spark --name arrow-spark --class org.apache.spark.examples.SparkPi --deploy-mode cluster local:///opt/spark/examples/jars/spark-examples_2.12-3.5.3.jar 100. Error code is: 1. Kubernetes spark exit code is: None
Last spark-submit output:
at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1435)
at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:545)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:484)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1429)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:918)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:176)
at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.handle(AbstractNioChannel.java:445)
at io.netty.channel.nio.NioIoHandler$DefaultNioRegistration.handle(NioIoHandler.java:388)
at io.netty.channel.nio.NioIoHandler.processSelectedKey(NioIoHandler.java:596)
at io.netty.channel.nio.NioIoHandler.processSelectedKeysOptimized(NioIoHandler.java:571)
at io.netty.channel.nio.NioIoHandler.processSelectedKeys(NioIoHandler.java:512)
at io.netty.channel.nio.NioIoHandler.run(NioIoHandler.java:484)
at io.netty.channel.SingleThreadIoEventLoop.runIo(SingleThreadIoEventLoop.java:225)
at io.netty.channel.SingleThreadIoEventLoop.run(SingleThreadIoEventLoop.java:196)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:1195)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:840)
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1593 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 195 in wrapper
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 2178 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 195 in wrapper
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 2136 in _run_execute_callable
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 445 in wrapper
File "/opt/airflow/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py", line 292 in execute
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/resumablejobmixin.py", line 200 in execute_resumable
File "/opt/airflow/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py", line 309 in submit_job
File "/opt/airflow/providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py", line 797 in submit
```
---
* 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] Anchor spark-submit log trail on actual exception instead of lines [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #70513:
URL: https://github.com/apache/airflow/pull/70513#discussion_r3657368999
##########
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:
Reverted, matches the original assertion now.
##########
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:
Trimmed to one line
--
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] Anchor spark-submit log trail on actual exception instead of lines [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #70513:
URL: https://github.com/apache/airflow/pull/70513#discussion_r3657325687
##########
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:
Verified against Spark 3.5.3 source before applying: `PythonRunner.scala` throws `SparkUserAppException` on a nonzero Python exit, and `SparkSubmit's` `ClassNotFoundException` handler does the same for a bad --class both caught by doSubmit's override and exited via exitFn(), so neither ever prints Exception in thread. Fair catch. Applied as suggested.
--
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] Anchor spark-submit log trail on actual exception instead of lines [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Anchor spark-submit log trail on actual exception instead of lines [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh merged PR #70513:
URL: https://github.com/apache/airflow/pull/70513
--
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] Anchor spark-submit log trail on actual exception instead of lines [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #70513:
URL: https://github.com/apache/airflow/pull/70513#issuecomment-5091798153
All handled in 74da146c1b
--
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] Anchor spark-submit log trail on actual exception instead of lines [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #70513:
URL: https://github.com/apache/airflow/pull/70513#discussion_r3657365829
##########
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:
Default reverted to `deque(maxlen=20)` as per comment above and added a test to actually exercise the 500-line cap once the anchor fires and confirms it caps and evicts from the front like any bounded deque, rather than being an untested 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