You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by fe...@apache.org on 2022/12/12 04:33:20 UTC
[incubator-kyuubi] branch master updated: [KYUUBI #3923] Add dedicated batch session idle timeout and prevent the batch submission operation timeout
This is an automated email from the ASF dual-hosted git repository.
feiwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
The following commit(s) were added to refs/heads/master by this push:
new fab235ebf [KYUUBI #3923] Add dedicated batch session idle timeout and prevent the batch submission operation timeout
fab235ebf is described below
commit fab235ebfc2b4f94892ecb7689e7f21e2b4002d8
Author: fwang12 <fw...@ebay.com>
AuthorDate: Mon Dec 12 12:33:12 2022 +0800
[KYUUBI #3923] Add dedicated batch session idle timeout and prevent the batch submission operation timeout
### _Why are the changes needed?_
Prevent batch session & operation closed by kyuubi because of timeout
### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
Closes #3923 from turboFei/batch_idle.
Closes #3923
76b0fe16 [fwang12] use val
310119b0 [fwang12] refactor
61ce2036 [fwang12] use session timeout
52fb0cc4 [fwang12] 1.6.2
7819f2091 [fwang12] save
69ffa316 [fwang12] check batch session timeout
dd210329 [fwang12] add batch op timeout
e304ddb4 [fwang12] never idle batch session and never timeout batch op
Authored-by: fwang12 <fw...@ebay.com>
Signed-off-by: fwang12 <fw...@ebay.com>
---
docs/deployment/settings.md | 1 +
.../src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala | 5 +++++
.../scala/org/apache/kyuubi/session/AbstractSession.scala | 2 ++
.../src/main/scala/org/apache/kyuubi/session/Session.scala | 1 +
.../scala/org/apache/kyuubi/session/SessionManager.scala | 5 ++---
.../org/apache/kyuubi/operation/BatchJobSubmission.scala | 2 ++
.../org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala | 12 ++++++++++++
7 files changed, 25 insertions(+), 3 deletions(-)
diff --git a/docs/deployment/settings.md b/docs/deployment/settings.md
index 16729869e..154bf6ba9 100644
--- a/docs/deployment/settings.md
+++ b/docs/deployment/settings.md
@@ -173,6 +173,7 @@ Key | Default | Meaning | Type | Since
kyuubi.batch.application.check.interval|PT5S|The interval to check batch job application information.|duration|1.6.0
kyuubi.batch.application.starvation.timeout|PT3M|Threshold above which to warn batch application may be starved.|duration|1.7.0
kyuubi.batch.conf.ignore.list||A comma separated list of ignored keys for batch conf. If the batch conf contains any of them, the key and the corresponding value will be removed silently during batch job submission. Note that this rule is for server-side protection defined via administrators to prevent some essential configs from tampering. You can also pre-define some config for batch job submission with prefix: kyuubi.batchConf.[batchType]. For example, you can pre-define `spark.master [...]
+kyuubi.batch.session.idle.timeout|PT6H|Batch session idle timeout, it will be closed when it's not accessed for this duration|duration|1.6.2
### Credentials
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
index 9a687fd8a..224c4d5ec 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
@@ -1189,6 +1189,11 @@ object KyuubiConf {
.version("1.2.0")
.fallbackConf(SESSION_TIMEOUT)
+ val BATCH_SESSION_IDLE_TIMEOUT: ConfigEntry[Long] = buildConf("kyuubi.batch.session.idle.timeout")
+ .doc("Batch session idle timeout, it will be closed when it's not accessed for this duration")
+ .version("1.6.2")
+ .fallbackConf(SESSION_IDLE_TIMEOUT)
+
val ENGINE_CHECK_INTERVAL: ConfigEntry[Long] = buildConf("kyuubi.session.engine.check.interval")
.doc("The check interval for engine timeout")
.version("1.0.0")
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
index 96a4ee4a6..0588d6702 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
@@ -53,6 +53,8 @@ abstract class AbstractSession(
if (lastIdleTime > 0) System.currentTimeMillis() - _lastIdleTime else 0
}
+ override val sessionIdleTimeoutThreshold: Long = sessionManager.getConf.get(SESSION_IDLE_TIMEOUT)
+
val normalizedConf: Map[String, String] = sessionManager.validateAndNormalizeConf(conf)
override lazy val name: Option[String] = normalizedConf.get(SESSION_NAME.key)
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala
index b5af13f7d..e45f93e36 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala
@@ -38,6 +38,7 @@ trait Session {
def lastAccessTime: Long
def lastIdleTime: Long
def getNoOperationTime: Long
+ def sessionIdleTimeoutThreshold: Long
def sessionManager: SessionManager
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/SessionManager.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/SessionManager.scala
index 26b4dae9c..b2eaba9bc 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/SessionManager.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/SessionManager.scala
@@ -290,15 +290,14 @@ abstract class SessionManager(name: String) extends CompositeService(name) {
private def startTimeoutChecker(): Unit = {
val interval = conf.get(SESSION_CHECK_INTERVAL)
- val timeout = conf.get(SESSION_IDLE_TIMEOUT)
val checkTask = new Runnable {
override def run(): Unit = {
val current = System.currentTimeMillis
if (!shutdown) {
for (session <- handleToSession.values().asScala) {
- if (session.lastAccessTime + timeout <= current &&
- session.getNoOperationTime > timeout) {
+ if (session.lastAccessTime + session.sessionIdleTimeoutThreshold <= current &&
+ session.getNoOperationTime > session.sessionIdleTimeoutThreshold) {
try {
closeSession(session.handle)
} catch {
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
index 2d8d63caa..6c9d5e3d3 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/operation/BatchJobSubmission.scala
@@ -337,6 +337,8 @@ class BatchJobSubmission(
override def cancel(): Unit = {
throw new IllegalStateException("Use close instead.")
}
+
+ override def isTimedOut: Boolean = false
}
object BatchJobSubmission {
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala
index 9aa642714..feccc8ee0 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiBatchSessionImpl.scala
@@ -59,6 +59,18 @@ class KyuubiBatchSessionImpl(
override def createTime: Long = recoveryMetadata.map(_.createTime).getOrElse(super.createTime)
+ override def getNoOperationTime: Long = {
+ if (batchJobSubmissionOp != null && !OperationState.isTerminal(
+ batchJobSubmissionOp.getStatus.state)) {
+ 0L
+ } else {
+ super.getNoOperationTime
+ }
+ }
+
+ override val sessionIdleTimeoutThreshold: Long =
+ sessionManager.getConf.get(KyuubiConf.BATCH_SESSION_IDLE_TIMEOUT)
+
// TODO: Support batch conf advisor
override val normalizedConf: Map[String, String] = {
sessionConf.getBatchConf(batchRequest.getBatchType) ++