You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2024/09/05 00:12:06 UTC
(spark) branch master updated: [SPARK-49478][CONNECT] Handle null metrics in ConnectProgressExecutionListener
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 9f2312d1e5bc [SPARK-49478][CONNECT] Handle null metrics in ConnectProgressExecutionListener
9f2312d1e5bc is described below
commit 9f2312d1e5bcc5300fe62477f77b3c8106238f2e
Author: Davin Tjong <da...@databricks.com>
AuthorDate: Thu Sep 5 09:11:54 2024 +0900
[SPARK-49478][CONNECT] Handle null metrics in ConnectProgressExecutionListener
### What changes were proposed in this pull request?
Handling null `TaskMetrics` in `ConnectProgressExecutionListenerSuite` by reporting 0 `inputBytesRead` on null.
### Why are the changes needed?
On task end, `TaskMetrics` may be `null`, as in the case of task failure (see [here](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala#L83)). This can cause NPEs for failed tasks with null metrics.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added a new test for task done with `null` metrics.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #47944 from davintjong-db/connect-progress-listener-null-metrics.
Lead-authored-by: Davin Tjong <da...@databricks.com>
Co-authored-by: Davin Tjong <10...@users.noreply.github.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../ConnectProgressExecutionListener.scala | 4 ++-
.../ConnectProgressExecutionListenerSuite.scala | 29 +++++++++++++++++++---
2 files changed, 28 insertions(+), 5 deletions(-)
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListener.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListener.scala
index a1881765a416..72c77fd033d7 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListener.scala
+++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListener.scala
@@ -144,7 +144,9 @@ private[connect] class ConnectProgressExecutionListener extends SparkListener wi
tracker.stages.get(taskEnd.stageId).foreach { stage =>
stage.update { i =>
i.completedTasks += 1
- i.inputBytesRead += taskEnd.taskMetrics.inputMetrics.bytesRead
+ i.inputBytesRead += Option(taskEnd.taskMetrics)
+ .map(_.inputMetrics.bytesRead)
+ .getOrElse(0L)
}
}
// This should never become negative, simply reset to zero if it does.
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListenerSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListenerSuite.scala
index 7c1b9362425d..df5df23e7750 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListenerSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ConnectProgressExecutionListenerSuite.scala
@@ -79,11 +79,16 @@ class ConnectProgressExecutionListenerSuite extends SparkFunSuite with MockitoSu
}
}
- test("taskDone") {
+ def testTaskDone(metricsPopulated: Boolean): Unit = {
val listener = new ConnectProgressExecutionListener
listener.registerJobTag(testTag)
listener.onJobStart(testJobStart)
+ val metricsOrNull = if (metricsPopulated) {
+ testStage1Task1Metrics
+ } else {
+ null
+ }
// Finish the tasks
val taskEnd = SparkListenerTaskEnd(
1,
@@ -92,7 +97,7 @@ class ConnectProgressExecutionListenerSuite extends SparkFunSuite with MockitoSu
Success,
testStage1Task1,
testStage1Task1ExecutorMetrics,
- testStage1Task1Metrics)
+ metricsOrNull)
val t = listener.trackedTags(testTag)
var yielded = false
@@ -117,7 +122,11 @@ class ConnectProgressExecutionListenerSuite extends SparkFunSuite with MockitoSu
assert(stages.map(_.numTasks).sum == 2)
assert(stages.map(_.completedTasks).sum == 1)
assert(stages.size == 2)
- assert(stages.map(_.inputBytesRead).sum == 500)
+ if (metricsPopulated) {
+ assert(stages.map(_.inputBytesRead).sum == 500)
+ } else {
+ assert(stages.map(_.inputBytesRead).sum == 0)
+ }
assert(
stages
.map(_.completed match {
@@ -140,7 +149,11 @@ class ConnectProgressExecutionListenerSuite extends SparkFunSuite with MockitoSu
assert(stages.map(_.numTasks).sum == 2)
assert(stages.map(_.completedTasks).sum == 1)
assert(stages.size == 2)
- assert(stages.map(_.inputBytesRead).sum == 500)
+ if (metricsPopulated) {
+ assert(stages.map(_.inputBytesRead).sum == 500)
+ } else {
+ assert(stages.map(_.inputBytesRead).sum == 0)
+ }
assert(
stages
.map(_.completed match {
@@ -153,4 +166,12 @@ class ConnectProgressExecutionListenerSuite extends SparkFunSuite with MockitoSu
assert(yielded, "Must updated with results")
}
+ test("taskDone - populated metrics") {
+ testTaskDone(metricsPopulated = true)
+ }
+
+ test("taskDone - null metrics") {
+ testTaskDone(metricsPopulated = false)
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org