[CARMEL-7550][CARMEL-4895] ExternalCatalogListener make SparkSession leak (apache#222)

Author: wakun

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -253,6 +253,10 @@ class SessionCatalog(
   private val listener = new ExternalCatalogListener(this)
   liveBus.map(_.addToExternalCatalogQueue(listener))
 
+  def removeCatalogListener(): Unit = {
+    liveBus.foreach(_.removeListener(listener))
+  }
+
   /** This method provides a way to get a cached plan. */
   def getCachedPlan(t: QualifiedTableName, c: Callable[(LogicalPlan, Int)]): LogicalPlan = {
     tableRelationCache.get(t, c)._1

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala

@@ -122,6 +122,7 @@ private[hive] class SparkSQLSessionManager(hiveServer: HiveServer2, sqlContext:
     ctx.sparkSession.scratchSessionPath.foreach { p =>
       SparkHadoopUtil.deletePath(p, ctx.sparkSession.sparkContext.hadoopConfiguration)
     }
+    ctx.sparkSession.sessionState.catalog.removeCatalogListener()
     super.closeSession(sessionHandle)
     sparkSqlOperationManager.sessionToContexts.remove(sessionHandle)
     AnalyticsTaskSchedulerImpl.activeSessions.decrementAndGet()