[KYUUBI #6204] Fix kyuubi session limiter leak when opening session f…

…ailed

# 🔍 Description
## Issue References 🔗

This pull request fixes #

## Describe Your Solution 🔧

We found that, a user has no active sessions, but kyuubi said that the user reach the max limit sessions per user.
Now, we increase the session limiter for user when opening session and decrease it when closing session.

But if the user open session failed, it will not decrease the session limiter.

This pr fix session limiter leak issue when failed to open session.

Before open session, add session handle into sessionHandleMap, and invoke SessionManager::closeSession when failed to open session.

## Types of changes 🔖

- [ ] Bugfix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)

## Test Plan 🧪

#### Behavior Without This Pull Request ⚰️

#### Behavior With This Pull Request 🎉

#### Related Unit Tests

---

# Checklist 📝

- [x] This patch was not authored or co-authored using [Generative Tooling](https://www.apache.org/legal/generative-tooling.html)

**Be nice. Be informative.**

Closes #6204 from turboFei/limiter_leak.

Closes #6204

c0f2969 [Wang, Fei] refine
98fda94 [Wang, Fei] fix leak

Authored-by: Wang, Fei <[email protected]>
Signed-off-by: Wang, Fei <[email protected]>

kyuubi-common/src/main/scala/org/apache/kyuubi/session/SessionManager.scala

@@ -103,16 +103,16 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
       conf: Map[String, String]): SessionHandle = {
     info(s"Opening session for $user@$ipAddress")
     val session = createSession(protocol, user, password, ipAddress, conf)
+    val handle = session.handle
     try {
-      val handle = session.handle
-      session.open()
       setSession(handle, session)
+      session.open()
       logSessionCountInfo(session, "opened")
       handle
     } catch {
       case e: Exception =>
         try {
-          session.close()
+          closeSession(handle)
         } catch {
           case t: Throwable =>
             warn(s"Error closing session for $user client ip: $ipAddress", t)

kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionManager.scala

@@ -175,14 +175,14 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
     batchLimiter.foreach(_.increment(UserIpAddress(user, ipAddress)))
     val handle = batchSession.handle
     try {
-      batchSession.open()
       setSession(handle, batchSession)
+      batchSession.open()
       logSessionCountInfo(batchSession, "opened")
       handle
     } catch {
       case e: Exception =>
         try {
-          batchSession.close()
+          closeSession(handle)
         } catch {
           case t: Throwable =>
             warn(s"Error closing batch session[$handle] for $user client ip: $ipAddress", t)