[LIVY-1009] Add support for global TTL for sessions

## What changes were proposed in this pull request?

Added two new Livy configuration parameters as below:

```
Enabled to check whether TTL Livy sessions should be stopped.
livy.server.session.ttl-check = false

Time in milliseconds on how long Livy will wait before TTL is an inactive session.
Note that the inactive session could be busy running jobs.
livy.server.session.ttl = 8h
```

## How was this patch tested?

Read LIVY-1009 for details.

Author: nileshrathi345

conf/livy.conf.template

@@ -67,6 +67,13 @@
 # How long a finished session state should be kept in LivyServer for query.
 # livy.server.session.state-retain.sec = 600s
 
+# Enabled to check whether TTL Livy sessions should be stopped.
+# livy.server.session.ttl-check = false
+
+# Time in milliseconds on how long Livy will wait before TTL is an inactive session.
+# Note that the inactive session could be busy running jobs.
+# livy.server.session.ttl = 8h
+
 # If livy should impersonate the requesting users when creating a new session.
 # livy.impersonation.enabled = false
 

server/src/main/scala/org/apache/livy/LivyConf.scala

@@ -354,6 +354,13 @@ object LivyConf {
   // Max creating session in livyServer
   val SESSION_MAX_CREATION = Entry("livy.server.session.max-creation", 100)
 
+  // Enabled to check whether TTL Livy sessions should be stopped.
+  val SESSION_TTL_CHECK = Entry("livy.server.session.ttl-check", false)
+
+  // Time in milliseconds on how long Livy will wait before TTL is an inactive session.
+  // Note that the inactive session could be busy running jobs.
+  val SESSION_TTL = Entry("livy.server.session.ttl", "8h")
+
   val SESSION_ALLOW_CUSTOM_CLASSPATH = Entry("livy.server.session.allow-custom-classpath", false)
 
   val SPARK_MASTER = "spark.master"

server/src/main/scala/org/apache/livy/sessions/SessionManager.scala

@@ -83,6 +83,10 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
 
   private[this] final val sessionTimeout = livyConf.getTimeAsMs(LivyConf.SESSION_TIMEOUT)
 
+  private[this] final val sessionTtlCheck = livyConf.getBoolean(LivyConf.SESSION_TTL_CHECK)
+
+  private[this] final val sessionTtl = livyConf.getTimeAsMs(LivyConf.SESSION_TTL)
+
   private[this] final val sessionStateRetainedInSec =
     TimeUnit.MILLISECONDS.toNanos(livyConf.getTimeAsMs(LivyConf.SESSION_STATE_RETAIN_TIME))
 
@@ -178,9 +182,13 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
             if (currentTime - session.lastActivity > calculatedTimeout) {
               return true
             }
-            if (session.ttl.isDefined && session.startedOn.isDefined) {
-              calculatedTimeout = TimeUnit.MILLISECONDS.toNanos(
-                ClientConf.getTimeAsMs(session.ttl.get))
+            if (sessionTtlCheck && session.startedOn.isDefined) {
+              if (session.ttl.isDefined) {
+                calculatedTimeout = TimeUnit.MILLISECONDS.toNanos(
+                  ClientConf.getTimeAsMs(session.ttl.get))
+              } else {
+                calculatedTimeout = TimeUnit.MILLISECONDS.toNanos(sessionTtl)
+              }
               if (currentTime - session.startedOn.get > calculatedTimeout) {
                 return true
               }