save

metrics

pending

doc

limit

fix

Author: turboFei

docs/configuration/settings.md

@@ -83,7 +83,9 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
 | 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 the prefix: kyuubi.batchConf.[batchType]. For example, you can pre-define `spark.master` for the Spark batch job with key `kyuubi.batchConf.spark.spark.master`. | set      | 1.6.0  |
 | kyuubi.batch.extra.resource.file.max.size   | 0       | The maximum size in bytes of each uploaded extra resource file when creating batch. 0 or negative value means no limit.                                                                                                                                                                                                                                                                                                                                                                                                                                    | long     | 1.10.0 |
+| kyuubi.batch.pending.check.window           | PT24H   | The time window to check the batch pending max elapse time from metadata store.                                                                                                                                                                                                                                                                                                                                                                                                                                                                            | duration | 1.10.1 |
 | kyuubi.batch.resource.file.max.size         | 0       | The maximum size in bytes of the uploaded resource file when creating batch. 0 or negative value means no limit.                                                                                                                                                                                                                                                                                                                                                                                                                                           | long     | 1.10.0 |
+| kyuubi.batch.search.window                  | PT72H   | The time window to search the batch from metadata store.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | duration | 1.10.1 |
 | 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

docs/monitor/metrics.md

@@ -1930,6 +1930,22 @@ object KyuubiConf {
       .stringConf
       .createWithDefault("1")
 
+  val BATCH_PENDING_CHECK_WINDOW: ConfigEntry[Long] =
+    buildConf("kyuubi.batch.pending.check.window")
+      .doc("The time window to check the batch pending max elapse time from metadata store.")
+      .version("1.10.1")
+      .timeConf
+      .checkValue(_ > 0, "must be positive number")
+      .createWithDefault(Duration.ofDays(1).toMillis)
+
+  val BATCH_SEARCH_WINDOW: ConfigEntry[Long] =
+    buildConf("kyuubi.batch.search.window")
+      .doc("The time window to search the batch from metadata store.")
+      .version("1.10.1")
+      .timeConf
+      .checkValue(_ > 0, "must be positive number")
+      .createWithDefault(Duration.ofDays(3).toMillis)
+
   val SERVER_EXEC_POOL_SIZE: ConfigEntry[Int] =
     buildConf("kyuubi.backend.server.exec.pool.size")
       .doc("Number of threads in the operation execution thread pool of Kyuubi server")

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

@@ -64,6 +64,7 @@ object MetricsConstants {
   final val OPERATION_TOTAL: String = OPERATION + "total"
   final val OPERATION_STATE: String = OPERATION + "state"
   final val OPERATION_EXEC_TIME: String = OPERATION + "exec_time"
+  final val OPERATION_BATCH_PENDING_MAX_ELAPSE: String = OPERATION + "batch_pending_max_elapse"
 
   final private val BACKEND_SERVICE = KYUUBI + "backend_service."
   final val BS_FETCH_LOG_ROWS_RATE = BACKEND_SERVICE + "fetch_log_rows_rate"

kyuubi-metrics/src/main/scala/org/apache/kyuubi/metrics/MetricsConstants.scala

@@ -31,8 +31,12 @@ import org.eclipse.jetty.servlet.{ErrorPageErrorHandler, FilterHolder}
 import org.apache.kyuubi.{KyuubiException, Utils}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.metrics.MetricsConstants
+import org.apache.kyuubi.metrics.MetricsSystem
+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.metadata.api.MetadataFilter
 import org.apache.kyuubi.server.ui.{JettyServer, JettyUtils}
 import org.apache.kyuubi.service.{AbstractFrontendService, Serverable, Service, ServiceUtils}
 import org.apache.kyuubi.service.authentication.{AuthTypes, AuthUtils}
@@ -202,6 +206,15 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
     }
   }
 
+  private def getBatchPendingMaxElapse(): Long = {
+    val filter = MetadataFilter(
+      state = OperationState.PENDING.toString,
+      kyuubiInstance = connectionUrl,
+      createTime = System.currentTimeMillis() - conf.get(BATCH_PENDING_CHECK_WINDOW))
+    sessionManager.getBatchesFromMetadataStore(filter, 0, 1, desc = false, orderByKeyId = false)
+      .headOption.map { batch => System.currentTimeMillis() - batch.getCreateTime }.getOrElse(0L)
+  }
+
   def waitForServerStarted(): Unit = {
     // block until the HTTP server is started, otherwise, we may get
     // the wrong HTTP server port -1
@@ -220,6 +233,12 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
         isStarted.set(true)
         startBatchChecker()
         recoverBatchSessions()
+        MetricsSystem.tracing { ms =>
+          ms.registerGauge(
+            MetricsConstants.OPERATION_BATCH_PENDING_MAX_ELAPSE,
+            getBatchPendingMaxElapse,
+            0)
+        }
       } catch {
         case e: Exception => throw new KyuubiException(s"Cannot start $getName", e)
       }

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

@@ -67,6 +67,7 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
     fe.getConf.get(ENGINE_SECURITY_ENABLED)
   private lazy val resourceFileMaxSize = fe.getConf.get(BATCH_RESOURCE_FILE_MAX_SIZE)
   private lazy val extraResourceFileMaxSize = fe.getConf.get(BATCH_EXTRA_RESOURCE_FILE_MAX_SIZE)
+  private lazy val batchSearchWindow = fe.getConf.get(BATCH_SEARCH_WINDOW)
 
   private def batchV2Enabled(reqConf: Map[String, String]): Boolean = {
     fe.getConf.get(BATCH_SUBMITTER_ENABLED) &&
@@ -426,9 +427,11 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
       username = batchUser,
       state = batchState,
       requestName = batchName,
-      createTime = createTime,
+      createTime = math.max(createTime, System.currentTimeMillis() - batchSearchWindow),
       endTime = endTime)
-    val batches = sessionManager.getBatchesFromMetadataStore(filter, from, size, desc)
+    // order by key_id(primary key auto increment column) is slower
+    val batches =
+      sessionManager.getBatchesFromMetadataStore(filter, from, size, desc, orderByKeyId = false)
     new GetBatchesResponse(from, batches.size, batches.asJava)
   }
 

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -138,8 +138,14 @@ class MetadataManager extends AbstractService("MetadataManager") {
       filter: MetadataFilter,
       from: Int,
       size: Int,
-      desc: Boolean = false): Seq[Batch] = {
-    withMetadataRequestMetrics(_metadataStore.getMetadataList(filter, from, size, desc)).map(
+      desc: Boolean = false,
+      orderByKeyId: Boolean = true): Seq[Batch] = {
+    withMetadataRequestMetrics(_metadataStore.getMetadataList(
+      filter,
+      from,
+      size,
+      desc,
+      orderByKeyId)).map(
       buildBatch)
   }
 

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataManager.scala

@@ -58,13 +58,16 @@ trait MetadataStore extends Closeable {
    * @param from the metadata offset.
    * @param size the size to get.
    * @param desc the order of metadata list.
+   * @param orderByKeyId the result order by auto increment key id,
+   *                     which is stable but might slow.
    * @return selected metadata list.
    */
   def getMetadataList(
       filter: MetadataFilter,
       from: Int,
       size: Int,
-      desc: Boolean = false): Seq[Metadata]
+      desc: Boolean = false,
+      orderByKeyId: Boolean = true): Seq[Metadata]
 
   /**
    * Count the metadata list with filter conditions.

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/MetadataStore.scala

@@ -257,14 +257,15 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
       filter: MetadataFilter,
       from: Int,
       size: Int,
-      desc: Boolean = false): Seq[Metadata] = {
+      desc: Boolean = false,
+      orderByKeyId: Boolean = true): Seq[Metadata] = {
     val queryBuilder = new StringBuilder
     val params = ListBuffer[Any]()
     queryBuilder.append("SELECT ")
     queryBuilder.append(METADATA_COLUMNS)
     queryBuilder.append(s" FROM $METADATA_TABLE")
     queryBuilder.append(s" ${assembleWhereClause(filter, params)}")
-    queryBuilder.append(" ORDER BY key_id ")
+    queryBuilder.append(" ORDER BY ").append(if (orderByKeyId) "key_id " else "create_time ")
     queryBuilder.append(if (desc) "DESC " else "ASC ")
     queryBuilder.append(dialect.limitClause(size, from))
     val query = queryBuilder.toString

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStore.scala

@@ -280,8 +280,9 @@ class KyuubiSessionManager private (name: String) extends SessionManager(name) {
       filter: MetadataFilter,
       from: Int,
       size: Int,
-      desc: Boolean = false): Seq[Batch] = {
-    metadataManager.map(_.getBatches(filter, from, size, desc)).getOrElse(Seq.empty)
+      desc: Boolean = false,
+      orderByKeyId: Boolean = true): Seq[Batch] = {
+    metadataManager.map(_.getBatches(filter, from, size, desc, orderByKeyId)).getOrElse(Seq.empty)
   }
 
   def getBatchMetadata(batchId: String): Option[Metadata] = {