[KYUUBI #958] [TEST] Ensure two connections in user mode share the same engine

<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
  2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->

### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about the use case of it.
  2. If you fix a bug, you can clarify why it is a bug.
-->

Ensure changes like #935 do not break the rule for the semantics of engine sharing

If and only if we want engines to be pooling,  this test can be changed

### _How was this patch tested?_
- [x] 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/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #958 from yaooqinn/same.

Closes #958

e172cebf [Kent Yao] Ensure two connections in user mode share the same engine

Authored-by: Kent Yao <yao@apache.org>
Signed-off-by: Kent Yao <yao@apache.org>

kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.operation
 
+import org.scalatest.time.SpanSugar._
+
 import org.apache.kyuubi.WithKyuubiServer
 import org.apache.kyuubi.config.KyuubiConf
 
@@ -27,4 +29,30 @@ class KyuubiOperationPerUserSuite extends WithKyuubiServer with JDBCTests {
   override protected val conf: KyuubiConf = {
     KyuubiConf().set(KyuubiConf.ENGINE_SHARE_LEVEL, "user")
   }
+
+  test("ensure two connections in user mode share the same engine") {
+    var r1: String = null
+    var r2: String = null
+    new Thread {
+      override def run(): Unit = withJdbcStatement() { statement =>
+        val res = statement.executeQuery("set spark.app.name")
+        assert(res.next())
+        r1 = res.getString("value")
+      }
+    }.start()
+
+    new Thread {
+      override def run(): Unit = withJdbcStatement() { statement =>
+        val res = statement.executeQuery("set spark.app.name")
+        assert(res.next())
+        r2 = res.getString("value")
+      }
+    }.start()
+
+    eventually(timeout(120.seconds), interval(100.milliseconds)) {
+      assert(r1 != null && r2 != null)
+    }
+
+    assert(r1 === r2)
+  }
 }