[SPARK-37315][ML][TEST] Mitigate ConcurrentModificationException thrown from a test in MLEventSuite

[sarutak]

What changes were proposed in this pull request?

This PR is to mitigate ConcurrentModificationException sometimes thrown from a test.
Recently, I notice the exception is thrown from the following part of the test pipeline read/write events in MLEventSuite when Scala 2.13 is used.

events.map(JsonProtocol.sparkEventToJson).foreach { event =>
  assert(JsonProtocol.sparkEventFromJson(event).isInstanceOf[MLEvent])
}

We can also find this issue from the scheduled build.
https://github.com/apache/spark/runs/4196812399?check_suite_focus=true#step:9:17616

I think the root cause is the ArrayBuffer (events) is updated asynchronously by the following part.

private val listener: SparkListener = new SparkListener {
  override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
    case e: MLEvent => events.append(e)
    case _ =>
  }
}

You can easily reproduce this issue by applying the following diff to the commit hash 4d29bec.

diff --git a/mllib/src/test/scala/org/apache/spark/ml/MLEventsSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/MLEventsSuite.scala
index f2343b7a88..ff63639e00 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/MLEventsSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/MLEventsSuite.scala
@@ -42,7 +42,9 @@ class MLEventsSuite
   private val events = mutable.ArrayBuffer.empty[MLEvent]
   private val listener: SparkListener = new SparkListener {
     override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
-      case e: MLEvent => events.append(e)
+      case e: MLEvent =>
+        events.append(e)
+        Thread.sleep(500)
       case _ =>
     }
   }
@@ -235,11 +237,13 @@ class MLEventsSuite
       }
       // Test if they can be ser/de via JSON protocol.
       assert(events.nonEmpty)
-      events.map(JsonProtocol.sparkEventToJson).foreach { event =>
-        assert(JsonProtocol.sparkEventFromJson(event).isInstanceOf[MLEvent])
-      }
+        events.map { x =>
+          Thread.sleep(500)
+          JsonProtocol.sparkEventToJson(x)
+        }.foreach { event =>
+          assert(JsonProtocol.sparkEventFromJson(event).isInstanceOf[MLEvent])
+        }
       sc.listenerBus.waitUntilEmpty(timeoutMillis = 10000)
-
       events.clear()
       val pipelineReader = Pipeline.read
       assert(events.isEmpty)

This is a kind of race condition but I think we can mitigate by retrying.

Actually, I have never seen this issue when I used Scala 2.13.5 and recently we upgrade to 2.13.7.
Scala 2.13.7 includes an update to detect ConcurrentModificationException more precisely.
scala/scala#9786

Why are the changes needed?

For test stability.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

I manually modified the test code, inserting sleep like the diff shown above, and confirmed no ConcurrentModificationException is thrown.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/49659/

[SparkQA]

Test build #145190 has finished for PR 34583 at commit f2443e1.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/49659/

[dongjoon-hyun]

+1, LGTM. I also observed this flakiness. Thank you, @sarutak .
Merged to master for Apache Spark 3.3.

[sarutak]

Thank you @dongjoon-hyun !