mridulm commented on code in PR #59966:
URL: https://github.com/apache/spark/pull/59966#discussion_r4489762704
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala:
##########
@@ -599,6 +599,42 @@ abstract class SQLAppStatusListenerSuite extends
SharedSparkSession with JsonTes
assert(metrics(driverMetric2.id) === expectedValue2)
}
+ test("SPARK-60759: aggregateMetrics merges repeated driver updates of the
same accumulator") {
+ val statusStore = spark.sharedState.statusStore
+ val oldCount = statusStore.executionsList().size
+
+ val values = Seq(10L, 20L)
+ val physicalPlan =
MyPlanWithRepeatedDriverAccumUpdates(sqlContext.sparkContext, values)
+ val dummyQueryExecution = new QueryExecution(spark, LocalRelation()) {
+ override lazy val sparkPlan = physicalPlan
+ override lazy val executedPlan = physicalPlan
+ }
+
+ SQLExecution.withNewExecutionId(dummyQueryExecution) {
+ physicalPlan.execute().collect()
+ }
+
+ while (statusStore.executionsCount() < oldCount) {
+ Thread.sleep(100)
+ }
+
+ while (statusStore.executionsList().isEmpty ||
+ statusStore.executionsList().last.metricValues == null) {
+ Thread.sleep(100)
+ }
Review Comment:
Use `eventually` for this (and below) instead - see other tests which are
leveraging it.
Also, we can combine the while loop above here as well into a single
condition.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
