[
https://issues.apache.org/jira/browse/LIVY-547?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16741731#comment-16741731
]
Wang Zhefeng commented on LIVY-547:
-----------------------------------
[~Sandeep Nemuri] Hi Sandeep，I think your session is an interactiveSession and
Livy gc thread will recycle your session when your job consume more time than
livy.server.session.timeout。
the relavant code：
{code:java}
def expired(session: Session): Boolean = {
session.state match {
case s: FinishedSessionState =>
val currentTime = System.nanoTime()
currentTime - s.time > sessionStateRetainedInSec
case _ =>
if (!sessionTimeoutCheck) {
false
} else if (session.isInstanceOf[BatchSession]) {
false
} else {
val currentTime = System.nanoTime()
currentTime - session.lastActivity > sessionTimeout
}
}{code}
you can see in last else，session will be recycled when duration time exceed
sessionTimeout，and session.lastActivity will not update in state busy.
I think this can be modified as:
{code:java}
def expired(session: Session): Boolean = {
session.state match {
case s: FinishedSessionState =>
val currentTime = System.nanoTime()
currentTime - s.time > sessionStateRetainedInSec
case _ =>
if (!sessionTimeoutCheck) {
false
} else if (session.isInstanceOf[BatchSession]) {
false
} else if("busy".equals(session.state.state)) {
false
} else {
val currentTime = System.nanoTime()
currentTime - session.lastActivity > sessionTimeout
}
}
}{code}
> Livy kills session after livy.server.session.timeout even if the session is
> active
> ----------------------------------------------------------------------------------
>
> Key: LIVY-547
> URL: https://issues.apache.org/jira/browse/LIVY-547
> Project: Livy
> Issue Type: Bug
> Components: Server
> Reporter: Sandeep Nemuri
> Priority: Major
>
> Livy kills session after {{livy.server.session.timeout}} even if the session
> is active.
> Code that runs more than the {{livy.server.session.timeout}} with
> intermediate sleeps.
> {noformat}
> %pyspark
> import time
> import datetime
> import random
> def inside(p):
> x, y = random.random(), random.random()
> return x*x + y*y < 1
> NUM_SAMPLES=10
> count = sc.parallelize(xrange(0, NUM_SAMPLES)) \
> .filter(inside).count()
> print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES)
> print("waiting for 100 s")
> time.sleep(100)
> count = sc.parallelize(xrange(0, NUM_SAMPLES)) \
> .filter(inside).count()
> print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES)
> print("waiting for 200 s")
> time.sleep(200)
> count = sc.parallelize(xrange(0, NUM_SAMPLES)) \
> .filter(inside).count()
> print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES)
> print("waiting for 300 s1")
> time.sleep(300)
> count = sc.parallelize(xrange(0, NUM_SAMPLES)) \
> .filter(inside).count()
> print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES)
> print("waiting for 300 s2")
> time.sleep(300)
> count = sc.parallelize(xrange(0, NUM_SAMPLES)) \
> .filter(inside).count()
> print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES)
> print("waiting for 300 s3")
> time.sleep(300)
> count = sc.parallelize(xrange(0, NUM_SAMPLES)) \
> .filter(inside).count()
> print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES)
> print("waiting for 300 s4")
> time.sleep(300)
> count = sc.parallelize(xrange(0, NUM_SAMPLES)) \
> .filter(inside).count()
> print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES)
> {noformat}
> Livy log:
> {noformat}
> 19/01/07 17:38:59 INFO InteractiveSession: Interactive session 14 created
> [appid: application_1546711709239_0002, owner: zeppelin-hwc327, proxyUser:
> Some(admin), state: idle, kind: shared, info:
> {driverLogUrl=http://hwc327-node3.hogwarts-labs.com:8042/node/containerlogs/container_e18_1546711709239_0002_01_000001/admin,
>
> sparkUiUrl=http://hwc327-node2.hogwarts-labs.com:8088/proxy/application_1546711709239_0002/}]
> 19/01/07 17:52:46 INFO InteractiveSession: Stopping InteractiveSession 14...
> 19/01/07 17:52:56 WARN RSCClient: Exception while waiting for end session
> reply.
> java.util.concurrent.TimeoutException
> at io.netty.util.concurrent.AbstractFuture.get(AbstractFuture.java:49)
> at org.apache.livy.rsc.RSCClient.stop(RSCClient.java:223)
> at
> org.apache.livy.server.interactive.InteractiveSession$$anonfun$stopSession$1.apply(InteractiveSession.scala:471)
> at
> org.apache.livy.server.interactive.InteractiveSession$$anonfun$stopSession$1.apply(InteractiveSession.scala:471)
> at scala.Option.foreach(Option.scala:236)
> at
> org.apache.livy.server.interactive.InteractiveSession.stopSession(InteractiveSession.scala:471)
> at
> org.apache.livy.sessions.Session$$anonfun$stop$1.apply$mcV$sp(Session.scala:174)
> at
> org.apache.livy.sessions.Session$$anonfun$stop$1.apply(Session.scala:171)
> at
> org.apache.livy.sessions.Session$$anonfun$stop$1.apply(Session.scala:171)
> at
> scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
> at
> scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
> at
> scala.concurrent.impl.ExecutionContextImpl$$anon$3.exec(ExecutionContextImpl.scala:107)
> at
> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> at
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> 19/01/07 17:52:56 WARN RpcDispatcher: [ClientProtocol] Closing RPC channel
> with 1 outstanding RPCs.
> 19/01/07 17:52:56 WARN InteractiveSession: Failed to stop RSCDriver. Killing
> it...
> 19/01/07 17:52:56 INFO YarnClientImpl: Killed application
> application_1546711709239_0002
> 19/01/07 17:52:56 INFO InteractiveSession: Stopped InteractiveSession 14.
> {noformat}
> Below note indicates that the timeout applies to a inactive session.
> {code:java|title=LivyConf.scala}
> // How long will an inactive session be gc-ed.
> val SESSION_TIMEOUT = Entry("livy.server.session.timeout", "1h")
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
