| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.livy.thriftserver |
| |
| import java.lang.reflect.UndeclaredThrowableException |
| import java.security.PrivilegedExceptionAction |
| import java.util |
| import java.util.{Date, Map => JMap, UUID} |
| import java.util.concurrent.{ConcurrentHashMap, LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit} |
| import java.util.concurrent.atomic.AtomicLong |
| import java.util.function.BiFunction |
| |
| import scala.collection.JavaConverters._ |
| import scala.collection.mutable |
| import scala.concurrent.{Await, Future} |
| import scala.concurrent.ExecutionContext.Implicits.global |
| import scala.concurrent.duration.Duration |
| import scala.util.{Failure, Success, Try} |
| |
| import org.apache.hadoop.security.UserGroupInformation |
| import org.apache.hive.service.cli.{HiveSQLException, SessionHandle} |
| import org.apache.hive.service.rpc.thrift.TProtocolVersion |
| import org.apache.hive.service.server.ThreadFactoryWithGarbageCleanup |
| |
| import org.apache.livy.LivyConf |
| import org.apache.livy.Logging |
| import org.apache.livy.server.interactive.{CreateInteractiveRequest, InteractiveSession} |
| import org.apache.livy.sessions.Spark |
| import org.apache.livy.thriftserver.SessionStates._ |
| import org.apache.livy.thriftserver.rpc.RpcClient |
| import org.apache.livy.utils.LivySparkUtils |
| |
| |
| class LivyThriftSessionManager(val server: LivyThriftServer, val livyConf: LivyConf) |
| extends ThriftService(classOf[LivyThriftSessionManager].getName) with Logging { |
| |
| private[thriftserver] val operationManager = new LivyOperationManager(this) |
| private[thriftserver] val sessionHandleToLivySession = |
| new ConcurrentHashMap[SessionHandle, Future[InteractiveSession]]() |
| // A map which returns how many incoming connections are open for a Livy session. |
| // This map tracks only the sessions created by the Livy thriftserver and not those which have |
| // been created through the REST API, as those should not be stopped even though there are no |
| // more active connections. |
| private val managedLivySessionActiveUsers = new mutable.HashMap[Int, Int]() |
| |
| // Contains metadata about a session |
| private val sessionInfo = new ConcurrentHashMap[SessionHandle, SessionInfo]() |
| |
| // Map the number of incoming connections for IP, user. It is used in order to check |
| // that the configured limits are not exceeded. |
| private val connectionsCount = new ConcurrentHashMap[String, AtomicLong] |
| |
| // Timeout for a Spark session creation |
| private val maxSessionWait = Duration( |
| server.livyConf.getTimeAsMs(LivyConf.THRIFT_SESSION_CREATION_TIMEOUT), |
| scala.concurrent.duration.MILLISECONDS) |
| |
| // Flag indicating whether the Spark version being used supports the USE database statement |
| val supportUseDatabase: Boolean = { |
| val sparkVersion = server.livyConf.get(LivyConf.LIVY_SPARK_VERSION) |
| val (sparkMajorVersion, _) = LivySparkUtils.formatSparkVersion(sparkVersion) |
| sparkMajorVersion > 1 || server.livyConf.getBoolean(LivyConf.ENABLE_HIVE_CONTEXT) |
| } |
| |
| // Configs from Hive |
| private val userLimit = livyConf.getInt(LivyConf.THRIFT_LIMIT_CONNECTIONS_PER_USER) |
| private val ipAddressLimit = livyConf.getInt(LivyConf.THRIFT_LIMIT_CONNECTIONS_PER_IPADDRESS) |
| private val userIpAddressLimit = |
| livyConf.getInt(LivyConf.THRIFT_LIMIT_CONNECTIONS_PER_USER_IPADDRESS) |
| private val checkInterval = livyConf.getTimeAsMs(LivyConf.THRIFT_SESSION_CHECK_INTERVAL) |
| private val sessionTimeout = livyConf.getTimeAsMs(LivyConf.THRIFT_IDLE_SESSION_TIMEOUT) |
| private val checkOperation = livyConf.getBoolean(LivyConf.THRIFT_IDLE_SESSION_CHECK_OPERATION) |
| |
| private var backgroundOperationPool: ThreadPoolExecutor = _ |
| |
| def getLivySession(sessionHandle: SessionHandle): InteractiveSession = { |
| val future = sessionHandleToLivySession.get(sessionHandle) |
| assert(future != null, s"Looking for not existing session: $sessionHandle.") |
| |
| if (!future.isCompleted) { |
| Try(Await.result(future, maxSessionWait)) match { |
| case Success(session) => session |
| case Failure(e) => throw Option(e.getCause).getOrElse(e) |
| } |
| } else { |
| future.value match { |
| case Some(Success(session)) => session |
| case Some(Failure(e)) => throw Option(e.getCause).getOrElse(e) |
| case None => throw new RuntimeException("Future cannot be None when it is completed") |
| } |
| } |
| } |
| |
| def livySessionId(sessionHandle: SessionHandle): Option[Int] = { |
| sessionHandleToLivySession.get(sessionHandle).value.filter(_.isSuccess).map(_.get.id) |
| } |
| |
| def livySessionState(sessionHandle: SessionHandle): SessionStates = { |
| sessionHandleToLivySession.get(sessionHandle).value match { |
| case Some(Success(_)) => CREATION_SUCCESS |
| case Some(Failure(_)) => CREATION_FAILED |
| case None => CREATION_IN_PROGRESS |
| } |
| } |
| |
| def onLivySessionOpened(livySession: InteractiveSession): Unit = { |
| server.livySessionManager.register(livySession) |
| synchronized { |
| managedLivySessionActiveUsers += livySession.id -> 0 |
| } |
| } |
| |
| def onUserSessionClosed(sessionHandle: SessionHandle, livySession: InteractiveSession): Unit = { |
| val closeSession = synchronized[Boolean] { |
| managedLivySessionActiveUsers.get(livySession.id) match { |
| case Some(1) => |
| // it was the last user, so we can close the LivySession |
| managedLivySessionActiveUsers -= livySession.id |
| true |
| case Some(activeUsers) => |
| managedLivySessionActiveUsers(livySession.id) = activeUsers - 1 |
| false |
| case None => |
| // This case can happen when we don't track the number of active users because the session |
| // has not been created in the thriftserver (ie. it has been created in the REST API). |
| false |
| } |
| } |
| if (closeSession) { |
| server.livySessionManager.delete(livySession) |
| } else { |
| // We unregister the session only if we don't close it, as it is unnecessary in that case |
| val rpcClient = new RpcClient(livySession) |
| try { |
| rpcClient.executeUnregisterSession(sessionHandle).get() |
| } catch { |
| case e: Exception => warn(s"Unable to unregister session $sessionHandle", e) |
| } |
| } |
| } |
| |
| /** |
| * If the user specified an existing sessionId to use, the corresponding session is returned, |
| * otherwise a new session is created and returned. |
| */ |
| private def getOrCreateLivySession( |
| sessionHandle: SessionHandle, |
| sessionId: Option[Int], |
| username: String, |
| createLivySession: () => InteractiveSession): InteractiveSession = { |
| sessionId match { |
| case Some(id) => |
| server.livySessionManager.get(id) match { |
| case None => |
| warn(s"InteractiveSession $id doesn't exist.") |
| throw new IllegalArgumentException(s"Session $id doesn't exist.") |
| case Some(session) if !server.isAllowedToUse(username, session) => |
| warn(s"$username has no modify permissions to InteractiveSession $id.") |
| throw new IllegalAccessException( |
| s"$username is not allowed to use InteractiveSession $id.") |
| case Some(session) => |
| if (session.state.isActive) { |
| info(s"Reusing Session $id for $sessionHandle.") |
| session |
| } else { |
| warn(s"InteractiveSession $id is not active anymore.") |
| throw new IllegalArgumentException(s"Session $id is not active anymore.") |
| } |
| } |
| case None => |
| createLivySession() |
| } |
| } |
| |
| /** |
| * Performs the initialization of the new Thriftserver session: |
| * - adds the Livy thrifserver JAR to the Spark application; |
| * - register the new Thriftserver session in the Spark application; |
| * - runs the initialization statements; |
| */ |
| private def initSession( |
| sessionHandle: SessionHandle, |
| livySession: InteractiveSession, |
| initStatements: List[String]): Unit = { |
| val rpcClient = new RpcClient(livySession) |
| rpcClient.executeRegisterSession(sessionHandle).get() |
| initStatements.foreach { statement => |
| val statementId = UUID.randomUUID().toString |
| try { |
| rpcClient.executeSql(sessionHandle, statementId, statement).get() |
| } finally { |
| Try(rpcClient.cleanupStatement(sessionHandle, statementId).get()).failed.foreach { e => |
| error(s"Failed to close init operation $statementId", e) |
| } |
| } |
| } |
| } |
| |
| def openSession( |
| protocol: TProtocolVersion, |
| username: String, |
| password: String, |
| ipAddress: String, |
| sessionConf: JMap[String, String], |
| withImpersonation: Boolean, |
| delegationToken: String): SessionHandle = { |
| val sessionHandle = new SessionHandle(protocol) |
| incrementConnections(username, ipAddress, SessionInfo.getForwardedAddresses) |
| sessionInfo.put(sessionHandle, |
| new SessionInfo(username, ipAddress, SessionInfo.getForwardedAddresses, protocol)) |
| val (initStatements, createInteractiveRequest, sessionId) = |
| LivyThriftSessionManager.processSessionConf(sessionConf, supportUseDatabase) |
| val createLivySession = () => { |
| createInteractiveRequest.kind = Spark |
| val newSession = InteractiveSession.create( |
| server.livySessionManager.nextId(), |
| createInteractiveRequest.name, |
| username, |
| None, |
| server.livyConf, |
| server.accessManager, |
| createInteractiveRequest, |
| server.sessionStore, |
| None, |
| None) |
| onLivySessionOpened(newSession) |
| newSession |
| } |
| val futureLivySession = Future { |
| val livyServiceUGI = UserGroupInformation.getCurrentUser |
| var livySession: InteractiveSession = null |
| try { |
| livyServiceUGI.doAs(new PrivilegedExceptionAction[InteractiveSession] { |
| override def run(): InteractiveSession = { |
| livySession = |
| getOrCreateLivySession(sessionHandle, sessionId, username, createLivySession) |
| synchronized { |
| managedLivySessionActiveUsers.get(livySession.id).foreach { numUsers => |
| managedLivySessionActiveUsers(livySession.id) = numUsers + 1 |
| } |
| } |
| initSession(sessionHandle, livySession, initStatements) |
| livySession |
| } |
| }) |
| } catch { |
| case e: UndeclaredThrowableException => |
| throw new ThriftSessionCreationException(Option(livySession), e.getCause) |
| case e: Throwable => |
| throw new ThriftSessionCreationException(Option(livySession), e) |
| } |
| } |
| sessionHandleToLivySession.put(sessionHandle, futureLivySession) |
| sessionHandle |
| } |
| |
| def closeSession(sessionHandle: SessionHandle): Unit = { |
| val removedSession = sessionHandleToLivySession.remove(sessionHandle) |
| val removedSessionInfo = sessionInfo.remove(sessionHandle) |
| try { |
| removedSession.value match { |
| case Some(Success(interactiveSession)) => |
| onUserSessionClosed(sessionHandle, interactiveSession) |
| case Some(Failure(e: ThriftSessionCreationException)) => |
| e.livySession.foreach(onUserSessionClosed(sessionHandle, _)) |
| case None => |
| removedSession.onComplete { |
| case Success(interactiveSession) => |
| onUserSessionClosed(sessionHandle, interactiveSession) |
| case Failure(e: ThriftSessionCreationException) => |
| e.livySession.foreach(onUserSessionClosed(sessionHandle, _)) |
| } |
| case _ => // We should never get here |
| } |
| } finally { |
| decrementConnections(removedSessionInfo) |
| } |
| } |
| |
| // Taken from Hive |
| override def init(livyConf: LivyConf): Unit = { |
| createBackgroundOperationPool(livyConf) |
| info("Connections limit are user: {} ipaddress: {} user-ipaddress: {}", |
| userLimit, ipAddressLimit, userIpAddressLimit) |
| super.init(livyConf) |
| } |
| |
| // Taken from Hive |
| private def createBackgroundOperationPool(livyConf: LivyConf): Unit = { |
| val poolSize = livyConf.getInt(LivyConf.THRIFT_ASYNC_EXEC_THREADS) |
| info("HiveServer2: Background operation thread pool size: " + poolSize) |
| val poolQueueSize = livyConf.getInt(LivyConf.THRIFT_ASYNC_EXEC_WAIT_QUEUE_SIZE) |
| info("HiveServer2: Background operation thread wait queue size: " + poolQueueSize) |
| val keepAliveTime = livyConf.getTimeAsMs(LivyConf.THRIFT_ASYNC_EXEC_KEEPALIVE_TIME) / 1000 |
| info(s"HiveServer2: Background operation thread keepalive time: $keepAliveTime seconds") |
| // Create a thread pool with #poolSize threads |
| // Threads terminate when they are idle for more than the keepAliveTime |
| // A bounded blocking queue is used to queue incoming operations, if #operations > poolSize |
| val threadPoolName = "LivyServer2-Background-Pool" |
| val queue = new LinkedBlockingQueue[Runnable](poolQueueSize) |
| backgroundOperationPool = new ThreadPoolExecutor( |
| poolSize, |
| poolSize, |
| keepAliveTime, |
| TimeUnit.SECONDS, |
| queue, |
| new ThreadFactoryWithGarbageCleanup(threadPoolName)) |
| backgroundOperationPool.allowCoreThreadTimeOut(true) |
| } |
| |
| // Taken from Hive |
| override def start(): Unit = { |
| super.start() |
| if (checkInterval > 0) startTimeoutChecker() |
| } |
| |
| private val timeoutCheckerLock: Object = new Object |
| @volatile private var shutdown: Boolean = false |
| |
| // Taken from Hive |
| private def startTimeoutChecker(): Unit = { |
| val interval: Long = Math.max(checkInterval, 3000L) |
| // minimum 3 seconds |
| val timeoutChecker: Runnable = new Runnable() { |
| override def run(): Unit = { |
| sleepFor(interval) |
| while (!shutdown) { |
| val current: Long = System.currentTimeMillis |
| val iterator = sessionHandleToLivySession.entrySet().iterator() |
| while (iterator.hasNext && ! shutdown) { |
| val entry = iterator.next() |
| val sessionHandle = entry.getKey |
| entry.getValue.value.flatMap(_.toOption).foreach { livySession => |
| |
| if (sessionTimeout > 0 && livySession.lastActivity + sessionTimeout <= current && |
| (!checkOperation || getNoOperationTime(sessionHandle) > sessionTimeout)) { |
| warn(s"Session $sessionHandle is Timed-out (last access : " + |
| new Date(livySession.lastActivity) + ") and will be closed") |
| try { |
| closeSession(sessionHandle) |
| } catch { |
| case e: HiveSQLException => |
| warn(s"Exception is thrown closing session $sessionHandle", e) |
| } |
| } else { |
| val operations = operationManager.getTimedOutOperations(sessionHandle) |
| if (operations.nonEmpty) { |
| operations.foreach { op => |
| try { |
| warn(s"Operation ${op.opHandle} is timed-out and will be closed") |
| operationManager.closeOperation(op.opHandle) |
| } catch { |
| case e: Exception => |
| warn("Exception is thrown closing timed-out operation: " + op.opHandle, e) |
| } |
| } |
| } |
| } |
| } |
| } |
| sleepFor(interval) |
| } |
| } |
| |
| private def sleepFor(interval: Long): Unit = { |
| timeoutCheckerLock.synchronized { |
| try { |
| timeoutCheckerLock.wait(interval) |
| } catch { |
| case e: InterruptedException => |
| // Ignore, and break. |
| } |
| } |
| } |
| } |
| backgroundOperationPool.execute(timeoutChecker) |
| } |
| |
| // Taken from Hive |
| private def shutdownTimeoutChecker(): Unit = { |
| shutdown = true |
| timeoutCheckerLock.synchronized { timeoutCheckerLock.notify() } |
| } |
| |
| // Taken from Hive |
| override def stop(): Unit = { |
| super.stop() |
| shutdownTimeoutChecker() |
| if (backgroundOperationPool != null) { |
| backgroundOperationPool.shutdown() |
| val timeout = livyConf.getTimeAsMs(LivyConf.THRIFT_ASYNC_EXEC_SHUTDOWN_TIMEOUT) / 1000 |
| try { |
| backgroundOperationPool.awaitTermination(timeout, TimeUnit.SECONDS) |
| } catch { |
| case e: InterruptedException => |
| warn(s"THRIFT_ASYNC_EXEC_SHUTDOWN_TIMEOUT = $timeout seconds has been exceeded. " + |
| "RUNNING background operations will be shut down", e) |
| } |
| backgroundOperationPool = null |
| } |
| } |
| |
| private def incrementConnectionsCount(key: String): Long = { |
| // Increment the count, returning the new value. |
| val count = connectionsCount.putIfAbsent(key, new AtomicLong(1L)) |
| if (count != null) { |
| count.incrementAndGet() |
| } else { |
| 1L |
| } |
| } |
| |
| private def decrementConnectionsCount(key: String): Unit = { |
| // Decrement the count. Remove the entry if the count reaches zero. |
| val decrementCount = new BiFunction[String, AtomicLong, AtomicLong] { |
| override def apply(unused: String, count: AtomicLong): AtomicLong = { |
| val countValue = count.decrementAndGet() |
| debug(s"Connection count for $key is $countValue") |
| if (countValue == 0L) { |
| null |
| } else { |
| count |
| } |
| } |
| } |
| connectionsCount.computeIfPresent(key, decrementCount) |
| } |
| |
| // Taken from Hive |
| private def getOriginClientIpAddress(ipAddress: String, forwardedAddresses: util.List[String]) = { |
| if (forwardedAddresses == null || forwardedAddresses.isEmpty) { |
| ipAddress |
| } else { |
| // order of forwarded ips per X-Forwarded-For http spec (client, proxy1, proxy2) |
| forwardedAddresses.get(0) |
| } |
| } |
| |
| private def logAndThrowException(msg: String): Unit = { |
| error(msg) |
| throw new HiveSQLException(msg) |
| } |
| |
| // Visible for testing |
| @throws[HiveSQLException] |
| private[thriftserver] def incrementConnections( |
| username: String, |
| ipAddress: String, |
| forwardedAddresses: util.List[String]): Unit = { |
| val clientIpAddress: String = getOriginClientIpAddress(ipAddress, forwardedAddresses) |
| val userAndAddress = username + ":" + clientIpAddress |
| val trackUser = trackConnectionsPerUser(username) |
| val trackIpAddress = trackConnectionsPerIpAddress(clientIpAddress) |
| val trackUserIpAddress = trackConnectionsPerUserIpAddress(username, clientIpAddress) |
| |
| // Optimistically increment the counts while getting them to check for violations. |
| // If any limit has been exceeded, we won't be going ahead with the connection, |
| // so decrement all counts that have been incremented. |
| if (trackUser) { |
| val userCount = incrementConnectionsCount(username) |
| if (userCount > userLimit) { |
| decrementConnectionsCount(username) |
| logAndThrowException("Connection limit per user reached " + |
| s"(user: $username limit: $userLimit)") |
| } |
| } |
| if (trackIpAddress) { |
| val ipAddressCount = incrementConnectionsCount(clientIpAddress) |
| if (ipAddressCount > ipAddressLimit) { |
| if (trackUser) decrementConnectionsCount(username) |
| decrementConnectionsCount(clientIpAddress) |
| logAndThrowException("Connection limit per ipaddress reached " + |
| s"(ipaddress: $clientIpAddress limit: $ipAddressLimit)") |
| } |
| } |
| if (trackUserIpAddress) { |
| val userIpAddressCount = incrementConnectionsCount(userAndAddress) |
| if (userIpAddressCount > userIpAddressLimit) { |
| if (trackUser) decrementConnectionsCount(username) |
| if (trackIpAddress) decrementConnectionsCount(clientIpAddress) |
| decrementConnectionsCount(userAndAddress) |
| logAndThrowException("Connection limit per user:ipaddress reached " + |
| s"(user:ipaddress: $userAndAddress limit: $userIpAddressLimit)") |
| } |
| } |
| } |
| |
| // Taken from Hive |
| private def trackConnectionsPerUserIpAddress(username: String, ipAddress: String): Boolean = { |
| userIpAddressLimit > 0 && username != null && !username.isEmpty && ipAddress != null && |
| !ipAddress.isEmpty |
| } |
| |
| // Taken from Hive |
| private def trackConnectionsPerIpAddress(ipAddress: String): Boolean = { |
| ipAddressLimit > 0 && ipAddress != null && !ipAddress.isEmpty |
| } |
| |
| // Taken from Hive |
| private def trackConnectionsPerUser(username: String): Boolean = { |
| userLimit > 0 && username != null && !username.isEmpty |
| } |
| |
| private def decrementConnections(sessionInfo: SessionInfo): Unit = { |
| val username = sessionInfo.username |
| val clientIpAddress = getOriginClientIpAddress( |
| sessionInfo.ipAddress, sessionInfo.forwardedAddresses) |
| if (trackConnectionsPerUser(username)) { |
| decrementConnectionsCount(username) |
| } |
| if (trackConnectionsPerIpAddress(clientIpAddress)) { |
| decrementConnectionsCount(clientIpAddress) |
| } |
| if (trackConnectionsPerUserIpAddress(username, clientIpAddress)) { |
| decrementConnectionsCount(username + ":" + clientIpAddress) |
| } |
| } |
| |
| def submitBackgroundOperation(r: Runnable): util.concurrent.Future[_] = { |
| backgroundOperationPool.submit(r) |
| } |
| |
| def getNoOperationTime(sessionHandle: SessionHandle): Long = { |
| if (operationManager.getOperations(sessionHandle).isEmpty) { |
| System.currentTimeMillis() - getLivySession(sessionHandle).lastActivity |
| } else { |
| 0 |
| } |
| } |
| |
| def getSessions: Set[SessionHandle] = { |
| sessionInfo.keySet().asScala.toSet |
| } |
| |
| def getSessionInfo(sessionHandle: SessionHandle): SessionInfo = { |
| sessionInfo.get(sessionHandle) |
| } |
| } |
| |
| object LivyThriftSessionManager extends Logging { |
| // Users can explicitly set the Livy connection id they want to connect to using this hiveconf |
| // variable |
| private val livySessionIdConfigKey = "set:hiveconf:livy.server.sessionId" |
| private val livySessionConfRegexp = "set:hiveconf:livy.session.conf.(.*)".r |
| private val hiveVarPattern = "set:hivevar:(.*)".r |
| |
| private def convertConfValueToInt(key: String, value: String) = { |
| val res = Try(value.toInt) |
| if (res.isFailure) { |
| warn(s"Ignoring $key = $value as it is not a valid integer") |
| None |
| } else { |
| Some(res.get) |
| } |
| } |
| |
| private def processSessionConf( |
| sessionConf: JMap[String, String], |
| supportUseDatabase: Boolean): (List[String], CreateInteractiveRequest, Option[Int]) = { |
| if (null != sessionConf && !sessionConf.isEmpty) { |
| val statements = new mutable.ListBuffer[String] |
| val extraLivyConf = new mutable.ListBuffer[(String, String)] |
| val createInteractiveRequest = new CreateInteractiveRequest |
| sessionConf.asScala.foreach { |
| case (key, value) => |
| key match { |
| case v if v.startsWith("use:") && supportUseDatabase => |
| statements += s"use $value" |
| // Process session configs for Livy session creation request |
| case "set:hiveconf:livy.session.driverMemory" => |
| createInteractiveRequest.driverMemory = Some(value) |
| case "set:hiveconf:livy.session.driverCores" => |
| createInteractiveRequest.driverCores = convertConfValueToInt(key, value) |
| case "set:hiveconf:livy.session.executorMemory" => |
| createInteractiveRequest.executorMemory = Some(value) |
| case "set:hiveconf:livy.session.executorCores" => |
| createInteractiveRequest.executorCores = convertConfValueToInt(key, value) |
| case "set:hiveconf:livy.session.numExecutors" => |
| createInteractiveRequest.numExecutors = convertConfValueToInt(key, value) |
| case "set:hiveconf:livy.session.queue" => |
| createInteractiveRequest.queue = Some(value) |
| case "set:hiveconf:livy.session.name" => |
| createInteractiveRequest.name = Some(value) |
| case "set:hiveconf:livy.session.heartbeatTimeoutInSecond" => |
| convertConfValueToInt(key, value).foreach { heartbeatTimeoutInSecond => |
| createInteractiveRequest.heartbeatTimeoutInSecond = heartbeatTimeoutInSecond |
| } |
| case livySessionConfRegexp(livyConfKey) => extraLivyConf += (livyConfKey -> value) |
| // set the hivevars specified by the user |
| case hiveVarPattern(confKey) => statements += s"set hivevar:${confKey.trim}=$value" |
| case _ if key == livySessionIdConfigKey => // Ignore it, we handle it later |
| case _ => |
| info(s"Ignoring key: $key = '$value'") |
| } |
| } |
| createInteractiveRequest.conf = extraLivyConf.toMap |
| val sessionId = Option(sessionConf.get(livySessionIdConfigKey)).flatMap { id => |
| val res = Try(id.toInt) |
| if (res.isFailure) { |
| warn(s"Ignoring $livySessionIdConfigKey=$id as it is not an int.") |
| None |
| } else { |
| Some(res.get) |
| } |
| } |
| (statements.toList, createInteractiveRequest, sessionId) |
| } else { |
| (List(), new CreateInteractiveRequest, None) |
| } |
| } |
| } |
| |
| /** |
| * Exception which happened during the session creation and/or initialization. It contains the |
| * `livySession` (if it was created) where the error occurred and the `cause` of the error. |
| */ |
| class ThriftSessionCreationException(val livySession: Option[InteractiveSession], cause: Throwable) |
| extends Exception(cause) |
