Skip to content

Commit 46346dd

Browse files
turboFeiturboFei
committed
[KYUUBI #7226] Support to wait the batch recovery appliction submission to throttle the load on the system
### Why are the changes needed? Support to wait the batch recovery appliction submission to throttle the load on the system. Add a new config to control it Whether a metadata recovery task should wait for its corresponding engine submission to complete before finishing. All recovery tasks are submitted to a fixed thread pool controlled by kyuubi.metadata.recovery.threads. If true, a task blocks until the engine submission is done, helping throttle the load on the system if kyuubi.session.engine.startup.waitCompletion is false. If false, the task returns immediately after opening the session without waiting. Close #7226 ### How was this patch tested? GA. ### Was this patch authored or co-authored using generative AI tooling? No. Closes #7262 from turboFei/recover_concurrent. Closes #7226 ea6282d [Wang, Fei] config 2b0403d [Wang, Fei] refine docs b5c5101 [Wang, Fei] refine f6b510c [Wang, Fei] 1.10.3 b892c71 [Wang, Fei] Support to wait the batch recovery appliction submission to throttle the load on the system c4740dc [Wang, Fei] conf Authored-by: Wang, Fei <[email protected]> Signed-off-by: Wang, Fei <[email protected]> (cherry picked from commit 572cef8) Signed-off-by: Wang, Fei <[email protected]>
1 parent 49eda80 commit 46346dd

File tree

3 files changed

+28
-1
lines changed

3 files changed

+28
-1
lines changed

‎docs/configuration/settings.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -384,6 +384,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
384384
| kyuubi.metadata.cleaner.interval | PT30M | The interval to check and clean expired metadata. | duration | 1.6.0 |
385385
| kyuubi.metadata.max.age | PT72H | The maximum age of metadata, the metadata exceeding the age will be cleaned. | duration | 1.6.0 |
386386
| kyuubi.metadata.recovery.threads | 10 | The number of threads for recovery from the metadata store when the Kyuubi server restarts. | int | 1.6.0 |
387+
| kyuubi.metadata.recovery.waitEngineSubmission | false | Whether a metadata recovery task should wait for its corresponding engine submission to complete before finishing. All recovery tasks are submitted to a fixed thread pool controlled by kyuubi.metadata.recovery.threads. If true, a task blocks until the engine submission is done, helping throttle the load on the system if kyuubi.session.engine.startup.waitCompletion is false. If false, the task returns immediately after opening the session without waiting. | boolean | 1.10.3 |
387388
| kyuubi.metadata.request.async.retry.enabled | true | Whether to retry in async when metadata request failed. When true, return success response immediately even the metadata request failed, and schedule it in background until success, to tolerate long-time metadata store outages w/o blocking the submission request. | boolean | 1.7.0 |
388389
| kyuubi.metadata.request.async.retry.queue.size | 65536 | The maximum queue size for buffering metadata requests in memory when the external metadata storage is down. Requests will be dropped if the queue exceeds. Only take affect when kyuubi.metadata.request.async.retry.enabled is `true`. | int | 1.6.0 |
389390
| kyuubi.metadata.request.async.retry.threads | 10 | Number of threads in the metadata request async retry manager thread pool. Only take affect when kyuubi.metadata.request.async.retry.enabled is `true`. | int | 1.6.0 |

‎kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2066,6 +2066,19 @@ object KyuubiConf {
20662066
.intConf
20672067
.createWithDefault(10)
20682068

2069+
val METADATA_RECOVERY_WAIT_ENGINE_SUBMISSION: ConfigEntry[Boolean] =
2070+
buildConf("kyuubi.metadata.recovery.waitEngineSubmission")
2071+
.serverOnly
2072+
.doc("Whether a metadata recovery task should wait for its corresponding engine " +
2073+
"submission to complete before finishing. All recovery tasks are submitted to a fixed " +
2074+
s"thread pool controlled by ${METADATA_RECOVERY_THREADS.key}. If true, a task blocks " +
2075+
"until the engine submission is done, helping throttle the load on the system " +
2076+
s"if ${SESSION_ENGINE_STARTUP_WAIT_COMPLETION.key} is false. " +
2077+
"If false, the task returns immediately after opening the session without waiting.")
2078+
.version("1.10.3")
2079+
.booleanConf
2080+
.createWithDefault(false)
2081+
20692082
val METADATA_REQUEST_RETRY_INTERVAL: ConfigEntry[Long] =
20702083
buildConf("kyuubi.metadata.request.retry.interval")
20712084
.serverOnly

‎kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ import org.apache.kyuubi.config.KyuubiConf
3333
import org.apache.kyuubi.config.KyuubiConf._
3434
import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem}
3535
import org.apache.kyuubi.metrics.MetricsConstants.OPERATION_BATCH_PENDING_MAX_ELAPSE
36+
import org.apache.kyuubi.operation.OperationState
3637
import org.apache.kyuubi.server.api.v1.ApiRootResource
3738
import org.apache.kyuubi.server.http.authentication.{AuthenticationFilter, KyuubiHttpAuthenticationFactory}
3839
import org.apache.kyuubi.server.ui.{JettyServer, JettyUtils}
@@ -170,6 +171,7 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
170171
@VisibleForTesting
171172
private[kyuubi] def recoverBatchSessions(): Unit = {
172173
val recoveryNumThreads = conf.get(METADATA_RECOVERY_THREADS)
174+
val recoveryWaitEngineSubmission = conf.get(METADATA_RECOVERY_WAIT_ENGINE_SUBMISSION)
173175
val batchRecoveryExecutor =
174176
ThreadUtils.newDaemonFixedThreadPool(recoveryNumThreads, "batch-recovery-executor")
175177
try {
@@ -179,7 +181,18 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
179181
val batchId = batchSession.batchJobSubmissionOp.batchId
180182
try {
181183
val task: Future[Unit] = batchRecoveryExecutor.submit(() =>
182-
Utils.tryLogNonFatalError(sessionManager.openBatchSession(batchSession)))
184+
Utils.tryLogNonFatalError {
185+
sessionManager.openBatchSession(batchSession)
186+
if (recoveryWaitEngineSubmission) {
187+
info(s"Waiting for batch[$batchId] engine submission during recovery")
188+
val batchOp = batchSession.batchJobSubmissionOp
189+
while (batchSession.getSessionEvent.forall(_.exception.isEmpty) &&
190+
!batchOp.appStarted &&
191+
!OperationState.isTerminal(batchOp.getStatus.state)) {
192+
Thread.sleep(300)
193+
}
194+
}
195+
})
183196
Some(task -> batchId)
184197
} catch {
185198
case e: Throwable =>

0 commit comments

Comments
 (0)