diff options
| author | guowei2 <guowei2@asiainfo.com> | 2014-12-18 20:10:23 -0800 |
|---|---|---|
| committer | Michael Armbrust <michael@databricks.com> | 2014-12-18 20:10:23 -0800 |
| commit | 22ddb6e0338f4d101389a0655424a8fde6c4cff4 (patch) | |
| tree | 6d6f15457970d5f1db8c31cf30e149834981bca8 /sql/hive-thriftserver/src | |
| parent | b68bc6d2647f8a5caf8aa558e4115f9cc254f67c (diff) | |
| download | spark-22ddb6e0338f4d101389a0655424a8fde6c4cff4.tar.gz spark-22ddb6e0338f4d101389a0655424a8fde6c4cff4.tar.bz2 spark-22ddb6e0338f4d101389a0655424a8fde6c4cff4.zip | |
[SPARK-4756][SQL] FIX: sessionToActivePool grow infinitely, even as sessions expire
**sessionToActivePool** in **SparkSQLOperationManager** grow infinitely, even as sessions expire.
we should remove the pool value when the session closed, even though **sessionToActivePool** would not exist in all of sessions.
Author: guowei2 <guowei2@asiainfo.com>
Closes #3617 from guowei2/SPARK-4756 and squashes the following commits:
e9b97b8 [guowei2] fix compile bug with Shim12
cf0f521 [guowei2] Merge remote-tracking branch 'apache/master' into SPARK-4756
e070998 [guowei2] fix: remove active pool of the session when it expired
Diffstat (limited to 'sql/hive-thriftserver/src')
2 files changed, 9 insertions, 3 deletions
diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala index 6b3275b4ea..89e9ede726 100644 --- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala +++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala @@ -27,11 +27,14 @@ import org.apache.hive.service.cli.session.SessionManager import org.apache.spark.sql.hive.HiveContext import org.apache.spark.sql.hive.thriftserver.ReflectionUtils._ import org.apache.spark.sql.hive.thriftserver.server.SparkSQLOperationManager +import org.apache.hive.service.cli.SessionHandle private[hive] class SparkSQLSessionManager(hiveContext: HiveContext) extends SessionManager with ReflectedCompositeService { + private lazy val sparkSqlOperationManager = new SparkSQLOperationManager(hiveContext) + override def init(hiveConf: HiveConf) { setSuperField(this, "hiveConf", hiveConf) @@ -40,10 +43,14 @@ private[hive] class SparkSQLSessionManager(hiveContext: HiveContext) getAncestorField[Log](this, 3, "LOG").info( s"HiveServer2: Async execution pool size $backgroundPoolSize") - val sparkSqlOperationManager = new SparkSQLOperationManager(hiveContext) setSuperField(this, "operationManager", sparkSqlOperationManager) addService(sparkSqlOperationManager) initCompositeService(hiveConf) } + + override def closeSession(sessionHandle: SessionHandle) { + super.closeSession(sessionHandle) + sparkSqlOperationManager.sessionToActivePool -= sessionHandle + } } diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/server/SparkSQLOperationManager.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/server/SparkSQLOperationManager.scala index 99c4f46a82..9c0bf02391 100644 --- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/server/SparkSQLOperationManager.scala +++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/server/SparkSQLOperationManager.scala @@ -36,8 +36,7 @@ private[thriftserver] class SparkSQLOperationManager(hiveContext: HiveContext) val handleToOperation = ReflectionUtils .getSuperField[JMap[OperationHandle, Operation]](this, "handleToOperation") - // TODO: Currenlty this will grow infinitely, even as sessions expire - val sessionToActivePool = Map[HiveSession, String]() + val sessionToActivePool = Map[SessionHandle, String]() override def newExecuteStatementOperation( parentSession: HiveSession, |