Skip to content

Commit b600264

Browse files
baibaichenCopilot
baibaichen
and
Copilot
committed
[GLUTEN-11550][UT] Add debug testGluten for JobTagging CI flaky test
Exclude original 'Tags set from session are prefixed with session UUID' test and add a testGluten version with logError diagnostics to debug the CI-only 'head of empty list' failure. The testGluten logs: - Main/Future thread identity and managedJobTags state - InheritableThreadLocal tag inheritance status - Actual job tags observed via SparkListener - cancelJobsWithTagWithFuture result Local run passes (5/5). CI failure root cause still under investigation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 6ef014c commit b600264

4 files changed

Lines changed: 188 additions & 2 deletions

File tree

‎gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -816,6 +816,7 @@ class VeloxTestSettings extends BackendTestSettings {
816816
// TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure
817817
enableSuite[GlutenSparkSessionBuilderSuite]
818818
enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite]
819+
.exclude("Tags set from session are prefixed with session UUID")
819820
enableSuite[GlutenTPCDSCollationQueryTestSuite]
820821
enableSuite[GlutenTPCDSModifiedPlanStabilitySuite]
821822
enableSuite[GlutenTPCDSModifiedPlanStabilityWithStatsSuite]

‎gluten-ut/spark40/src/test/scala/org/apache/spark/sql/GlutenSparkSessionJobTaggingAndCancellationSuite.scala‎

Lines changed: 93 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,98 @@
1616
*/
1717
package org.apache.spark.sql
1818

19+
import org.apache.spark.SparkContext
20+
import org.apache.spark.internal.Logging
21+
import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
22+
import org.apache.spark.sql.execution.SQLExecution
23+
import org.apache.spark.util.ThreadUtils
24+
25+
import java.util.concurrent.{Semaphore, TimeUnit}
26+
27+
import scala.concurrent.{ExecutionContext, Future}
28+
import scala.concurrent.duration._
29+
1930
class GlutenSparkSessionJobTaggingAndCancellationSuite
2031
extends SparkSessionJobTaggingAndCancellationSuite
21-
with GlutenTestSetWithSystemPropertyTrait {}
32+
with GlutenTestSetWithSystemPropertyTrait
33+
with Logging {
34+
35+
testGluten("GLUTEN-DEBUG Tags set from session are prefixed with session UUID") {
36+
sc = new SparkContext("local[2]", "test")
37+
val session = classic.SparkSession.builder().sparkContext(sc).getOrCreate()
38+
import session.implicits._
39+
40+
val mainThread = Thread.currentThread()
41+
logError(
42+
s"[GLUTEN-DEBUG] Main thread: ${mainThread.getName} " +
43+
s"(id=${mainThread.getId}, class=${mainThread.getClass.getName})")
44+
45+
val sem = new Semaphore(0)
46+
val jobTags = new java.util.concurrent.ConcurrentHashMap[Int, String]()
47+
sc.addSparkListener(new SparkListener {
48+
override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
49+
val tags = Option(jobStart.properties.getProperty(SparkContext.SPARK_JOB_TAGS))
50+
.getOrElse("<null>")
51+
jobTags.put(jobStart.jobId, tags)
52+
logError(s"[GLUTEN-DEBUG] onJobStart: jobId=${jobStart.jobId}, tags=$tags")
53+
sem.release()
54+
}
55+
})
56+
57+
session.addTag("one")
58+
val mainManagedTags = session.managedJobTags.get().toMap
59+
val mainThreadUuid = session.threadUuid.get()
60+
logError(s"[GLUTEN-DEBUG] Main thread managedJobTags: $mainManagedTags")
61+
logError(s"[GLUTEN-DEBUG] Main thread threadUuid: $mainThreadUuid")
62+
63+
Future {
64+
val futureThread = Thread.currentThread()
65+
val futureManagedTags = session.managedJobTags.get().toMap
66+
val futureThreadUuid = session.threadUuid.get()
67+
logError(
68+
s"[GLUTEN-DEBUG] Future thread: ${futureThread.getName} " +
69+
s"(id=${futureThread.getId}, class=${futureThread.getClass.getName})")
70+
logError(s"[GLUTEN-DEBUG] Future thread managedJobTags: $futureManagedTags")
71+
logError(s"[GLUTEN-DEBUG] Future thread threadUuid: $futureThreadUuid")
72+
logError(
73+
s"[GLUTEN-DEBUG] Future thread inherited tag 'one': " +
74+
s"${futureManagedTags.contains("one")}")
75+
76+
session.range(1, 10000).map { i => Thread.sleep(100); i }.count()
77+
}(ExecutionContext.global)
78+
79+
val acquired = sem.tryAcquire(1, 1, TimeUnit.MINUTES)
80+
logError(s"[GLUTEN-DEBUG] Semaphore acquired: $acquired")
81+
logError(s"[GLUTEN-DEBUG] All job tags observed: $jobTags")
82+
83+
val realTag = session.managedJobTags.get()("one")
84+
logError(s"[GLUTEN-DEBUG] Looking for tag: $realTag")
85+
86+
val activeJobsFuture =
87+
session.sparkContext.cancelJobsWithTagWithFuture(realTag, "reason")
88+
val cancelledJobs = ThreadUtils.awaitResult(activeJobsFuture, 60.seconds)
89+
logError(s"[GLUTEN-DEBUG] Cancelled jobs count: ${cancelledJobs.size}")
90+
cancelledJobs.foreach(job => logError(s"[GLUTEN-DEBUG] Cancelled job: ${job.jobId}"))
91+
92+
assert(
93+
cancelledJobs.nonEmpty,
94+
s"Expected at least one cancelled job. " +
95+
s"mainTags=$mainManagedTags, realTag=$realTag, observedJobTags=$jobTags")
96+
97+
val activeJob = cancelledJobs.head
98+
val actualTags = activeJob.properties
99+
.getProperty(SparkContext.SPARK_JOB_TAGS)
100+
.split(SparkContext.SPARK_JOB_TAGS_SEP)
101+
assert(
102+
actualTags.toSet == Set(
103+
session.sessionJobTag,
104+
s"${session.sessionJobTag}-thread-${session.threadUuid.get()}-one",
105+
SQLExecution.executionIdJobTag(
106+
session,
107+
activeJob.properties
108+
.get(SQLExecution.EXECUTION_ROOT_ID_KEY)
109+
.asInstanceOf[String]
110+
.toLong)
111+
))
112+
}
113+
}

‎gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -800,6 +800,7 @@ class VeloxTestSettings extends BackendTestSettings {
800800
// TODO: 4.x enableSuite[GlutenSetCommandSuite] // 1 failure
801801
enableSuite[GlutenSparkSessionBuilderSuite]
802802
enableSuite[GlutenSparkSessionJobTaggingAndCancellationSuite]
803+
.exclude("Tags set from session are prefixed with session UUID")
803804
enableSuite[GlutenTPCDSCollationQueryTestSuite]
804805
enableSuite[GlutenTPCDSModifiedPlanStabilitySuite]
805806
enableSuite[GlutenTPCDSModifiedPlanStabilityWithStatsSuite]

‎gluten-ut/spark41/src/test/scala/org/apache/spark/sql/GlutenSparkSessionJobTaggingAndCancellationSuite.scala‎

Lines changed: 93 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,98 @@
1616
*/
1717
package org.apache.spark.sql
1818

19+
import org.apache.spark.SparkContext
20+
import org.apache.spark.internal.Logging
21+
import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
22+
import org.apache.spark.sql.execution.SQLExecution
23+
import org.apache.spark.util.ThreadUtils
24+
25+
import java.util.concurrent.{Semaphore, TimeUnit}
26+
27+
import scala.concurrent.{ExecutionContext, Future}
28+
import scala.concurrent.duration._
29+
1930
class GlutenSparkSessionJobTaggingAndCancellationSuite
2031
extends SparkSessionJobTaggingAndCancellationSuite
21-
with GlutenTestSetWithSystemPropertyTrait {}
32+
with GlutenTestSetWithSystemPropertyTrait
33+
with Logging {
34+
35+
testGluten("GLUTEN-DEBUG Tags set from session are prefixed with session UUID") {
36+
sc = new SparkContext("local[2]", "test")
37+
val session = classic.SparkSession.builder().sparkContext(sc).getOrCreate()
38+
import session.implicits._
39+
40+
val mainThread = Thread.currentThread()
41+
logError(
42+
s"[GLUTEN-DEBUG] Main thread: ${mainThread.getName} " +
43+
s"(id=${mainThread.getId}, class=${mainThread.getClass.getName})")
44+
45+
val sem = new Semaphore(0)
46+
val jobTags = new java.util.concurrent.ConcurrentHashMap[Int, String]()
47+
sc.addSparkListener(new SparkListener {
48+
override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
49+
val tags = Option(jobStart.properties.getProperty(SparkContext.SPARK_JOB_TAGS))
50+
.getOrElse("<null>")
51+
jobTags.put(jobStart.jobId, tags)
52+
logError(s"[GLUTEN-DEBUG] onJobStart: jobId=${jobStart.jobId}, tags=$tags")
53+
sem.release()
54+
}
55+
})
56+
57+
session.addTag("one")
58+
val mainManagedTags = session.managedJobTags.get().toMap
59+
val mainThreadUuid = session.threadUuid.get()
60+
logError(s"[GLUTEN-DEBUG] Main thread managedJobTags: $mainManagedTags")
61+
logError(s"[GLUTEN-DEBUG] Main thread threadUuid: $mainThreadUuid")
62+
63+
Future {
64+
val futureThread = Thread.currentThread()
65+
val futureManagedTags = session.managedJobTags.get().toMap
66+
val futureThreadUuid = session.threadUuid.get()
67+
logError(
68+
s"[GLUTEN-DEBUG] Future thread: ${futureThread.getName} " +
69+
s"(id=${futureThread.getId}, class=${futureThread.getClass.getName})")
70+
logError(s"[GLUTEN-DEBUG] Future thread managedJobTags: $futureManagedTags")
71+
logError(s"[GLUTEN-DEBUG] Future thread threadUuid: $futureThreadUuid")
72+
logError(
73+
s"[GLUTEN-DEBUG] Future thread inherited tag 'one': " +
74+
s"${futureManagedTags.contains("one")}")
75+
76+
session.range(1, 10000).map { i => Thread.sleep(100); i }.count()
77+
}(ExecutionContext.global)
78+
79+
val acquired = sem.tryAcquire(1, 1, TimeUnit.MINUTES)
80+
logError(s"[GLUTEN-DEBUG] Semaphore acquired: $acquired")
81+
logError(s"[GLUTEN-DEBUG] All job tags observed: $jobTags")
82+
83+
val realTag = session.managedJobTags.get()("one")
84+
logError(s"[GLUTEN-DEBUG] Looking for tag: $realTag")
85+
86+
val activeJobsFuture =
87+
session.sparkContext.cancelJobsWithTagWithFuture(realTag, "reason")
88+
val cancelledJobs = ThreadUtils.awaitResult(activeJobsFuture, 60.seconds)
89+
logError(s"[GLUTEN-DEBUG] Cancelled jobs count: ${cancelledJobs.size}")
90+
cancelledJobs.foreach(job => logError(s"[GLUTEN-DEBUG] Cancelled job: ${job.jobId}"))
91+
92+
assert(
93+
cancelledJobs.nonEmpty,
94+
s"Expected at least one cancelled job. " +
95+
s"mainTags=$mainManagedTags, realTag=$realTag, observedJobTags=$jobTags")
96+
97+
val activeJob = cancelledJobs.head
98+
val actualTags = activeJob.properties
99+
.getProperty(SparkContext.SPARK_JOB_TAGS)
100+
.split(SparkContext.SPARK_JOB_TAGS_SEP)
101+
assert(
102+
actualTags.toSet == Set(
103+
session.sessionJobTag,
104+
s"${session.sessionJobTag}-thread-${session.threadUuid.get()}-one",
105+
SQLExecution.executionIdJobTag(
106+
session,
107+
activeJob.properties
108+
.get(SQLExecution.EXECUTION_ROOT_ID_KEY)
109+
.asInstanceOf[String]
110+
.toLong)
111+
))
112+
}
113+
}

0 commit comments

Comments
 (0)