Skip to content

Commit 572cef8

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]>
1 parent 05d0637 commit 572cef8

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
@@ -387,6 +387,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
387387
| kyuubi.metadata.cleaner.interval | PT30M | The interval to check and clean expired metadata. | duration | 1.6.0 |
388388
| kyuubi.metadata.max.age | PT72H | The maximum age of metadata, the metadata exceeding the age will be cleaned. | duration | 1.6.0 |
389389
| kyuubi.metadata.recovery.threads | 10 | The number of threads for recovery from the metadata store when the Kyuubi server restarts. | int | 1.6.0 |
390+
| 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 |
390391
| 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 |
391392
| 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 |
392393
| 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
@@ -2104,6 +2104,19 @@ object KyuubiConf {
21042104
.intConf
21052105
.createWithDefault(10)
21062106

2107+
val METADATA_RECOVERY_WAIT_ENGINE_SUBMISSION: ConfigEntry[Boolean] =
2108+
buildConf("kyuubi.metadata.recovery.waitEngineSubmission")
2109+
.serverOnly
2110+
.doc("Whether a metadata recovery task should wait for its corresponding engine " +
2111+
"submission to complete before finishing. All recovery tasks are submitted to a fixed " +
2112+
s"thread pool controlled by ${METADATA_RECOVERY_THREADS.key}. If true, a task blocks " +
2113+
"until the engine submission is done, helping throttle the load on the system " +
2114+
s"if ${SESSION_ENGINE_STARTUP_WAIT_COMPLETION.key} is false. " +
2115+
"If false, the task returns immediately after opening the session without waiting.")
2116+
.version("1.10.3")
2117+
.booleanConf
2118+
.createWithDefault(false)
2119+
21072120
val METADATA_REQUEST_RETRY_INTERVAL: ConfigEntry[Long] =
21082121
buildConf("kyuubi.metadata.request.retry.interval")
21092122
.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
@@ -34,6 +34,7 @@ import org.apache.kyuubi.config.KyuubiConf
3434
import org.apache.kyuubi.config.KyuubiConf._
3535
import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem}
3636
import org.apache.kyuubi.metrics.MetricsConstants.OPERATION_BATCH_PENDING_MAX_ELAPSE
37+
import org.apache.kyuubi.operation.OperationState
3738
import org.apache.kyuubi.server.api.v1.ApiRootResource
3839
import org.apache.kyuubi.server.http.authentication.{AuthenticationFilter, KyuubiHttpAuthenticationFactory}
3940
import org.apache.kyuubi.server.ui.{JettyServer, JettyUtils}
@@ -181,6 +182,7 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
181182
@VisibleForTesting
182183
private[kyuubi] def recoverBatchSessions(): Unit = withBatchRecoveryLockRequired {
183184
val recoveryNumThreads = conf.get(METADATA_RECOVERY_THREADS)
185+
val recoveryWaitEngineSubmission = conf.get(METADATA_RECOVERY_WAIT_ENGINE_SUBMISSION)
184186
val batchRecoveryExecutor =
185187
ThreadUtils.newDaemonFixedThreadPool(recoveryNumThreads, "batch-recovery-executor")
186188
try {
@@ -190,7 +192,18 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
190192
val batchId = batchSession.batchJobSubmissionOp.batchId
191193
try {
192194
val task: Future[Unit] = batchRecoveryExecutor.submit(() =>
193-
Utils.tryLogNonFatalError(sessionManager.openBatchSession(batchSession)))
195+
Utils.tryLogNonFatalError {
196+
sessionManager.openBatchSession(batchSession)
197+
if (recoveryWaitEngineSubmission) {
198+
info(s"Waiting for batch[$batchId] engine submission during recovery")
199+
val batchOp = batchSession.batchJobSubmissionOp
200+
while (batchSession.getSessionEvent.forall(_.exception.isEmpty) &&
201+
!batchOp.appStarted &&
202+
!OperationState.isTerminal(batchOp.getStatus.state)) {
203+
Thread.sleep(300)
204+
}
205+
}
206+
})
194207
Some(task -> batchId)
195208
} catch {
196209
case e: Throwable =>

0 commit comments

Comments
 (0)