(uniffle) branch master updated: [#2487] fix(spark3): Fix NPE in UniffleListener (#2488)

[zuston]

This is an automated email from the ASF dual-hosted git repository.

zuston pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/uniffle.git


The following commit(s) were added to refs/heads/master by this push:
     new b7faef885 [#2487] fix(spark3): Fix NPE in UniffleListener (#2488)
b7faef885 is described below

commit b7faef8859dfe4995b26b60764e26e6b7449bdf5
Author: Zhen Wang <643348...@qq.com>
AuthorDate: Tue May 20 16:57:16 2025 +0800

    [#2487] fix(spark3): Fix NPE in UniffleListener (#2488)
    
    ### What changes were proposed in this pull request?
    
    fix npe
    
    ### Why are the changes needed?
    
    closes #2487
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    minor fix
---
 .../extension/src/main/scala/org/apache/spark/UniffleListener.scala   | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git 
a/client-spark/extension/src/main/scala/org/apache/spark/UniffleListener.scala 
b/client-spark/extension/src/main/scala/org/apache/spark/UniffleListener.scala
index 560007cbb..f290bd09a 100644
--- 
a/client-spark/extension/src/main/scala/org/apache/spark/UniffleListener.scala
+++ 
b/client-spark/extension/src/main/scala/org/apache/spark/UniffleListener.scala
@@ -60,8 +60,8 @@ class UniffleListener(conf: SparkConf, kvstore: 
ElementTrackingStore)
 
   override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
     this.mayUpdate(false)
-    if (taskEnd.taskMetrics.shuffleReadMetrics.recordsRead > 0
-      || taskEnd.taskMetrics.shuffleWriteMetrics.recordsWritten > 0) {
+    if (taskEnd.taskMetrics != null && 
(taskEnd.taskMetrics.shuffleReadMetrics.recordsRead > 0
+      || taskEnd.taskMetrics.shuffleWriteMetrics.recordsWritten > 0)) {
       totalTaskCpuTime.addAndGet(taskEnd.taskInfo.duration)
       
totalShuffleWriteTime.addAndGet(taskEnd.taskMetrics.shuffleWriteMetrics.writeTime
 / 1000000)
       
totalShuffleReadTime.addAndGet(taskEnd.taskMetrics.shuffleReadMetrics.fetchWaitTime)