[KYUUBI #2208] Fixed session close operator log session dir not deleted

### _Why are the changes needed?_
Operator log Session dir not deleted when the session is closed，
![image](https://user-images.githubusercontent.com/13195083/159858277-ee1b4391-3f87-41e9-906e-bc02f8ef4655.png)
As time goes on, more and more directories are created, which is a big risk

### _How was this patch tested?_
1、open session
![image](https://user-images.githubusercontent.com/13195083/159859323-221ffd93-4500-4236-8646-9253e3019983.png)
session operator dir
![image](https://user-images.githubusercontent.com/13195083/159859633-4bb3db5a-8d46-4e79-af3f-ad88cb1eb3d9.png)
2、Closing the session directory is deleted
![image](https://user-images.githubusercontent.com/13195083/159860262-adcd7f60-2838-456b-beef-6413c50433b7.png)

![image](https://user-images.githubusercontent.com/13195083/159860191-9defcd23-875b-4f03-8e98-9469dcba52f9.png)

Closes #2208 from winfys/master.

Closes #2208

500e8cf [winfys] Fixed session close operator log session dir not deleted
315d835 [winfys] Fixed session close operator log session dir not deleted
7f261a3 [winfys] Fixed session close operator log session dir not deleted
2dad761 [Fengyuanshen] Merge branch 'apache:master' into master
46725f0 [winfys] Fixed session close operator log session dir not deleted
cbcee43 [winfys] Fixed session close operator log session dir not deleted
43e1ec1 [winfys] Fixed session close operator log session dir not deleted

Lead-authored-by: winfys <yuanshen.feng@yingxiong.com>
Co-authored-by: Fengyuanshen <18211673094@163.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>
(cherry picked from commit a13a89e)
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: 2 people

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

@@ -89,10 +89,23 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
     if (session == null) {
       throw KyuubiSQLException(s"Invalid $sessionHandle")
     }
+    deleteOperationLogSessionDir(sessionHandle)
     info(s"$sessionHandle is closed, current opening sessions $getOpenSessionCount")
     session.close()
   }
 
+  private def deleteOperationLogSessionDir(sessionHandle: SessionHandle): Unit = {
+    _operationLogRoot.foreach(logRoot => {
+      val rootPath = Paths.get(logRoot, sessionHandle.identifier.toString)
+      try {
+        Files.deleteIfExists(rootPath)
+      } catch {
+        case e: IOException =>
+          error(s"Failed to delete session operation log directory ${rootPath.toString}", e)
+      }
+    })
+  }
+
   def getSession(sessionHandle: SessionHandle): Session = {
     val session = handleToSession.get(sessionHandle)
     if (session == null) {

kyuubi-common/src/test/scala/org/apache/kyuubi/operation/log/OperationLogSuite.scala

@@ -82,6 +82,9 @@ class OperationLogSuite extends KyuubiFunSuite {
 
     operationLog.close()
     assert(!Files.exists(logFile))
+    assert(Files.exists(Paths.get(operationLogRoot, sHandle.identifier.toString)))
+    sessionManager.closeSession(sHandle)
+    assert(!Files.exists(Paths.get(operationLogRoot, sHandle.identifier.toString)))
   }
 
   test("log divert appender") {