diff --git a/docs/configuration/settings.md b/docs/configuration/settings.md index ab1966cde62..bd45438e37e 100644 --- a/docs/configuration/settings.md +++ b/docs/configuration/settings.md @@ -387,6 +387,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co | kyuubi.metadata.cleaner.interval | PT30M | The interval to check and clean expired metadata. | duration | 1.6.0 | | kyuubi.metadata.max.age | PT72H | The maximum age of metadata, the metadata exceeding the age will be cleaned. | duration | 1.6.0 | | kyuubi.metadata.recovery.threads | 10 | The number of threads for recovery from the metadata store when the Kyuubi server restarts. | int | 1.6.0 | +| 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 | | 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 | | 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 | | 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 | 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 eb17e222ed8..c1fad17ec0a 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 @@ -2104,6 +2104,19 @@ object KyuubiConf { .intConf .createWithDefault(10) + val METADATA_RECOVERY_WAIT_ENGINE_SUBMISSION: ConfigEntry[Boolean] = + buildConf("kyuubi.metadata.recovery.waitEngineSubmission") + .serverOnly + .doc("Whether a metadata recovery task should wait for its corresponding engine " + + "submission to complete before finishing. All recovery tasks are submitted to a fixed " + + s"thread pool controlled by ${METADATA_RECOVERY_THREADS.key}. If true, a task blocks " + + "until the engine submission is done, helping throttle the load on the system " + + s"if ${SESSION_ENGINE_STARTUP_WAIT_COMPLETION.key} is false. " + + "If false, the task returns immediately after opening the session without waiting.") + .version("1.10.3") + .booleanConf + .createWithDefault(false) + val METADATA_REQUEST_RETRY_INTERVAL: ConfigEntry[Long] = buildConf("kyuubi.metadata.request.retry.interval") .serverOnly diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala index 787ac0b0473..8f9be59a54a 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala @@ -34,6 +34,7 @@ import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ import org.apache.kyuubi.metrics.{MetricsConstants, MetricsSystem} import org.apache.kyuubi.metrics.MetricsConstants.OPERATION_BATCH_PENDING_MAX_ELAPSE +import org.apache.kyuubi.operation.OperationState import org.apache.kyuubi.server.api.v1.ApiRootResource import org.apache.kyuubi.server.http.authentication.{AuthenticationFilter, KyuubiHttpAuthenticationFactory} import org.apache.kyuubi.server.ui.{JettyServer, JettyUtils} @@ -181,6 +182,7 @@ class KyuubiRestFrontendService(override val serverable: Serverable) @VisibleForTesting private[kyuubi] def recoverBatchSessions(): Unit = withBatchRecoveryLockRequired { val recoveryNumThreads = conf.get(METADATA_RECOVERY_THREADS) + val recoveryWaitEngineSubmission = conf.get(METADATA_RECOVERY_WAIT_ENGINE_SUBMISSION) val batchRecoveryExecutor = ThreadUtils.newDaemonFixedThreadPool(recoveryNumThreads, "batch-recovery-executor") try { @@ -190,7 +192,18 @@ class KyuubiRestFrontendService(override val serverable: Serverable) val batchId = batchSession.batchJobSubmissionOp.batchId try { val task: Future[Unit] = batchRecoveryExecutor.submit(() => - Utils.tryLogNonFatalError(sessionManager.openBatchSession(batchSession))) + Utils.tryLogNonFatalError { + sessionManager.openBatchSession(batchSession) + if (recoveryWaitEngineSubmission) { + info(s"Waiting for batch[$batchId] engine submission during recovery") + val batchOp = batchSession.batchJobSubmissionOp + while (batchSession.getSessionEvent.forall(_.exception.isEmpty) && + !batchOp.appStarted && + !OperationState.isTerminal(batchOp.getStatus.state)) { + Thread.sleep(300) + } + } + }) Some(task -> batchId) } catch { case e: Throwable =>