[spark] Stabilize row tracking concurrent tests (#8307)

Row tracking concurrent tests could fail on transient commit conflicts
and snapshot reads while merge and compaction run in parallel. The
previous retry logic was duplicated and, in one path, retried without a
bound.

This PR centralizes the retry helper for these concurrent tests, adds a
bounded retry with a small backoff, preserves the final exception on
failure, and narrows snapshot missing detection to require `Snapshot
file` and `does not exist` in the same throwable message.

Author: QuakeWang

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala

@@ -44,6 +44,43 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar
 
   import testImplicits._
 
+  private val MaxRetryAttempts = 20
+  private val RetryIntervalMillis = 10L
+
+  private def doWithRetry(doAction: () => Unit, retryableMessages: String*): Unit = {
+    var attempts = 0
+    while (true) {
+      try {
+        doAction.apply()
+        return
+      } catch {
+        case e: Exception =>
+          attempts += 1
+          if (!isRetryable(e, retryableMessages) || attempts >= MaxRetryAttempts) {
+            throw e
+          }
+          Thread.sleep(RetryIntervalMillis)
+      }
+    }
+  }
+
+  private def isRetryable(e: Throwable, retryableMessages: Seq[String]): Boolean = {
+    hasMessage(e, "Snapshot file", "does not exist") ||
+    retryableMessages.exists(message => hasMessage(e, message))
+  }
+
+  private def hasMessage(e: Throwable, fragments: String*): Boolean = {
+    var current = e
+    while (current != null) {
+      val message = current.getMessage
+      if (message != null && fragments.forall(message.contains)) {
+        return true
+      }
+      current = current.getCause
+    }
+    false
+  }
+
   test("Data Evolution: concurrent merge and compact") {
     withTable("s", "t") {
       sql(s"""
@@ -104,27 +141,16 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar
       Seq((1, 1, 1)).toDF("id", "b", "c").createOrReplaceTempView("s")
 
       def doMerge(): Unit = {
-        var success = false
-        while (!success) {
-          try {
-            sql(s"""
-                   |MERGE INTO t
-                   |USING s
-                   |ON t.id = s.id
-                   |WHEN MATCHED THEN
-                   |UPDATE SET t.id = s.id, t.b = s.b + t.b, t.c = s.c + t.c
-                   |""".stripMargin).collect()
-            success = true
-          } catch {
-            case e: Exception =>
-              if (
-                !e.getMessage.contains(
-                  "multiple 'MERGE INTO' operations have encountered conflicts")
-              ) {
-                throw e
-              }
-          }
-        }
+        doWithRetry(
+          () => sql(s"""
+                       |MERGE INTO t
+                       |USING s
+                       |ON t.id = s.id
+                       |WHEN MATCHED THEN
+                       |UPDATE SET t.id = s.id, t.b = s.b + t.b, t.c = s.c + t.c
+                       |""".stripMargin).collect(),
+          "multiple 'MERGE INTO' operations have encountered conflicts"
+        )
       }
 
       val mergeInto1 = Future {
@@ -159,25 +185,25 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar
 
       val mergeB = Future {
         for (_ <- 1 to 10) {
-          sql(s"""
-                 |MERGE INTO t
-                 |USING sb
-                 |ON t.id = sb.id
-                 |WHEN MATCHED THEN
-                 |UPDATE SET t.b = sb.b + t.b
-                 |""".stripMargin).collect()
+          doWithRetry(() => sql(s"""
+                                   |MERGE INTO t
+                                   |USING sb
+                                   |ON t.id = sb.id
+                                   |WHEN MATCHED THEN
+                                   |UPDATE SET t.b = sb.b + t.b
+                                   |""".stripMargin).collect())
         }
       }
 
       val mergeC = Future {
         for (_ <- 1 to 10) {
-          sql(s"""
-                 |MERGE INTO t
-                 |USING sc
-                 |ON t.id = sc.id
-                 |WHEN MATCHED THEN
-                 |UPDATE SET t.c = sc.c + t.c
-                 |""".stripMargin).collect()
+          doWithRetry(() => sql(s"""
+                                   |MERGE INTO t
+                                   |USING sc
+                                   |ON t.id = sc.id
+                                   |WHEN MATCHED THEN
+                                   |UPDATE SET t.c = sc.c + t.c
+                                   |""".stripMargin).collect())
         }
       }
 
@@ -199,33 +225,19 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar
       sql("INSERT INTO t VALUES (1, 0, 0)")
       Seq((1, 1, 1)).toDF("id", "b", "c").createOrReplaceTempView("s")
 
-      def doWithRetry(doAction: () => Unit): Unit = {
-        var success = false
-        while (!success) {
-          try {
-            doAction.apply()
-            success = true
-          } catch {
-            case e: Exception =>
-              if (
-                !e.getMessage.contains("multiple 'MERGE INTO' and 'COMPACT' operations")
-                && !e.getMessage.contains("Row ID existence conflict")
-              ) {
-                throw e
-              }
-          }
-        }
-      }
-
       val mergeInto = Future {
         for (i <- 1 to 10) {
-          doWithRetry(() => sql(s"""
-                                   |MERGE INTO t
-                                   |USING s
-                                   |ON t.id = s.id
-                                   |WHEN MATCHED THEN
-                                   |UPDATE SET t.id = s.id, t.b = s.b + t.b, t.c = s.c + t.c
-                                   |""".stripMargin).collect())
+          doWithRetry(
+            () => sql(s"""
+                         |MERGE INTO t
+                         |USING s
+                         |ON t.id = s.id
+                         |WHEN MATCHED THEN
+                         |UPDATE SET t.id = s.id, t.b = s.b + t.b, t.c = s.c + t.c
+                         |""".stripMargin).collect(),
+            "multiple 'MERGE INTO' and 'COMPACT' operations",
+            "Row ID existence conflict"
+          )
           if (i > 1) {
             sql(s"INSERT INTO t VALUES ($i, $i, $i)")
           }
@@ -244,7 +256,10 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar
           while (!canBeCompacted) {
             Thread.sleep(1)
           }
-          doWithRetry(() => sql("CALL sys.compact(table => 't')"))
+          doWithRetry(
+            () => sql("CALL sys.compact(table => 't')"),
+            "multiple 'MERGE INTO' and 'COMPACT' operations",
+            "Row ID existence conflict")
         }
       }