[KYUUBI #4865] Fix flaky test BackendServiceMetricSuite

### _Why are the changes needed?_

Wrap assertion w/ `eventually` to address the flaky test.

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.readthedocs.io/en/master/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #4865 from pan3793/flaky-test-2.

Closes #4865

d582d57 [Cheng Pan] Fix flaky test BackendServiceMetricSuite

Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

Author: pan3793

kyuubi-server/src/test/scala/org/apache/kyuubi/server/BackendServiceMetricSuite.scala

@@ -21,10 +21,12 @@ import java.nio.file.{Path, Paths}
 import java.time.Duration
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
 
 import org.apache.kyuubi.{Utils, WithKyuubiServer}
 import org.apache.kyuubi.config.KyuubiConf
-import org.apache.kyuubi.metrics.{MetricsConf, MetricsConstants}
+import org.apache.kyuubi.metrics.MetricsConf
+import org.apache.kyuubi.metrics.MetricsConstants._
 import org.apache.kyuubi.operation.HiveJDBCTestHelper
 
 class BackendServiceMetricSuite extends WithKyuubiServer with HiveJDBCTestHelper {
@@ -45,43 +47,40 @@ class BackendServiceMetricSuite extends WithKyuubiServer with HiveJDBCTestHelper
     withJdbcStatement() { statement =>
       statement.executeQuery("CREATE TABLE stu_test(id int, name string) USING parquet")
       statement.execute("insert into stu_test values(1, 'a'), (2, 'b'), (3, 'c')")
-      Thread.sleep(Duration.ofMillis(111).toMillis)
+      val logRows1 = eventually(timeout(10.seconds), interval(1.second)) {
+        val res = objMapper.readTree(Paths.get(reportPath.toString, "report.json").toFile)
+        assert(res.has("timers"))
+        val timer = res.get("timers")
+        assert(timer.get(BS_EXECUTE_STATEMENT).get("count").asInt() == 2)
+        assert(timer.get(BS_EXECUTE_STATEMENT).get("mean").asDouble() > 0)
 
-      val res1 = objMapper.readTree(Paths.get(reportPath.toString, "report.json").toFile)
-      assert(res1.has("timers"))
-      val timer1 = res1.get("timers")
-      assert(
-        timer1.get(MetricsConstants.BS_EXECUTE_STATEMENT).get("count").asInt() == 2)
-      assert(
-        timer1.get(MetricsConstants.BS_EXECUTE_STATEMENT).get("mean").asDouble() > 0)
-
-      assert(res1.has("meters"))
-      val meters1 = res1.get("meters")
-      val logRows1 = meters1.get(MetricsConstants.BS_FETCH_LOG_ROWS_RATE).get("count").asInt()
-      assert(logRows1 > 0)
+        assert(res.has("meters"))
+        val meters = res.get("meters")
+        val logRows = meters.get(BS_FETCH_LOG_ROWS_RATE).get("count").asInt()
+        assert(logRows > 0)
+        logRows
+      }
 
       statement.execute("select * from stu_test limit 2")
       statement.getResultSet.next()
-      Thread.sleep(Duration.ofMillis(111).toMillis)
-
-      val res2 = objMapper.readTree(Paths.get(reportPath.toString, "report.json").toFile)
-      val timer2 = res2.get("timers")
-      assert(
-        timer2.get(MetricsConstants.BS_OPEN_SESSION).get("count").asInt() == 1)
-      assert(
-        timer2.get(MetricsConstants.BS_OPEN_SESSION).get("min").asInt() > 0)
-      val execStatementNode2 = timer2.get(MetricsConstants.BS_EXECUTE_STATEMENT)
-      assert(execStatementNode2.get("count").asInt() == 3)
-      assert(
-        execStatementNode2.get("max").asDouble() >= execStatementNode2.get("mean").asDouble() &&
-          execStatementNode2.get("mean").asDouble() >= execStatementNode2.get("min").asDouble())
+      eventually(timeout(60.seconds), interval(1.second)) {
+        val res = objMapper.readTree(Paths.get(reportPath.toString, "report.json").toFile)
+        val timer = res.get("timers")
+        assert(timer.get(BS_OPEN_SESSION).get("count").asInt() == 1)
+        assert(timer.get(BS_OPEN_SESSION).get("min").asInt() > 0)
+        val execStatementNode = timer.get(BS_EXECUTE_STATEMENT)
+        assert(execStatementNode.get("count").asInt() == 3)
+        assert(
+          execStatementNode.get("max").asDouble() >= execStatementNode.get("mean").asDouble() &&
+            execStatementNode.get("mean").asDouble() >= execStatementNode.get("min").asDouble())
 
-      val meters2 =
-        objMapper.readTree(Paths.get(reportPath.toString, "report.json").toFile).get("meters")
-      assert(meters2.get(MetricsConstants.BS_FETCH_RESULT_ROWS_RATE).get("count").asInt() == 8)
-      assert(meters2.get(MetricsConstants.BS_FETCH_LOG_ROWS_RATE).get("count").asInt() >= logRows1)
+        val meters =
+          objMapper.readTree(Paths.get(reportPath.toString, "report.json").toFile).get("meters")
+        assert(meters.get(BS_FETCH_RESULT_ROWS_RATE).get("count").asInt() == 8)
+        assert(meters.get(BS_FETCH_LOG_ROWS_RATE).get("count").asInt() >= logRows1)
 
-      statement.executeQuery("DROP TABLE stu_test")
+        statement.executeQuery("DROP TABLE stu_test")
+      }
     }
   }
 }