saev

Author: turboFei

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

@@ -2086,7 +2086,7 @@ object KyuubiConf {
         "when there are too many expired metadata to be cleaned.")
       .version("1.11.0")
       .timeConf
-      .createWithDefault(Duration.ofMillis(100))
+      .createWithDefault(Duration.ofMillis(100).toMillis)
 
   val METADATA_CLEANER_BATCH_SIZE: ConfigEntry[Int] =
     buildConf("kyuubi.metadata.cleaner.batch.size")

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

@@ -17,11 +17,11 @@
 
 package org.apache.kyuubi.server.metadata
 
+import com.google.common.annotations.VisibleForTesting
+
 import java.util.concurrent.{ConcurrentHashMap, ThreadPoolExecutor, TimeUnit}
 import java.util.concurrent.atomic.AtomicInteger
-
 import scala.collection.JavaConverters._
-
 import org.apache.kyuubi.{KyuubiException, Logging}
 import org.apache.kyuubi.client.api.v1.dto.Batch
 import org.apache.kyuubi.config.KyuubiConf
@@ -240,27 +240,7 @@ class MetadataManager extends AbstractService("MetadataManager") {
     val batchInterval = conf.get(KyuubiConf.METADATA_CLEANER_BATCH_INTERVAL)
     val cleanerTask: Runnable = () => {
       try {
-        var needToCleanMetadata = true
-        var needToCleanKubernetesInfo = true
-
-        while (needToCleanMetadata || needToCleanKubernetesInfo) {
-          if (needToCleanMetadata) {
-            needToCleanMetadata =
-              withMetadataRequestMetrics(_metadataStore.cleanupMetadataByAge(
-                stateMaxAge,
-                batchSize)) >= batchSize
-          }
-          if (needToCleanKubernetesInfo) {
-            needToCleanKubernetesInfo =
-              withMetadataRequestMetrics(_metadataStore.cleanupKubernetesEngineInfoByAge(
-                stateMaxAge,
-                batchSize)) >= batchSize
-          }
-          if (needToCleanMetadata || needToCleanKubernetesInfo) {
-            info("Sleep for " + batchInterval + "ms before next metadata cleanup batch")
-            Thread.sleep(batchInterval)
-          }
-        }
+        cleanupMetadata(stateMaxAge, batchSize, batchInterval)
       } catch {
         case e: Throwable => error("Error cleaning up the metadata by age", e)
       }
@@ -274,6 +254,31 @@ class MetadataManager extends AbstractService("MetadataManager") {
       TimeUnit.MILLISECONDS)
   }
 
+  @VisibleForTesting
+  private[metadata] def cleanupMetadata(maxAge: Long, batchSize: Int, batchInterval: Long): Unit = {
+    var needToCleanMetadata = true
+    var needToCleanKubernetesInfo = true
+
+    while (needToCleanMetadata || needToCleanKubernetesInfo) {
+      if (needToCleanMetadata) {
+        needToCleanMetadata =
+          withMetadataRequestMetrics(_metadataStore.cleanupMetadataByAge(
+            maxAge,
+            batchSize)) >= batchSize
+      }
+      if (needToCleanKubernetesInfo) {
+        needToCleanKubernetesInfo =
+          withMetadataRequestMetrics(_metadataStore.cleanupKubernetesEngineInfoByAge(
+            maxAge,
+            batchSize)) >= batchSize
+      }
+      if (needToCleanMetadata || needToCleanKubernetesInfo) {
+        info("Sleep for " + batchInterval + "ms before next metadata cleanup batch")
+        Thread.sleep(batchInterval)
+      }
+    }
+  }
+
   def addMetadataRetryRequest(request: MetadataRequest): Unit = {
     val maxRequestsAsyncRetryRefs: Int =
       conf.get(KyuubiConf.METADATA_REQUEST_ASYNC_RETRY_QUEUE_SIZE)

kyuubi-server/src/test/scala/org/apache/kyuubi/server/metadata/MetadataManagerSuite.scala

@@ -212,8 +212,22 @@ class MetadataManagerSuite extends KyuubiFunSuite {
       f(metadataManager)
     } finally {
       metadataManager.getBatches(MetadataFilter(), 0, Int.MaxValue).foreach { batch =>
-        metadataManager.cleanupMetadataById(batch.getId)
+        // close the batch if not ended
+        if (batch.getEndTime == 0) {
+          metadataManager.updateMetadata(Metadata(
+            identifier = batch.getId,
+            state = OperationState.CLOSED.toString,
+            endTime = System.currentTimeMillis()), false)
+        }
       }
+
+      metadataManager.cleanupMetadata(0, 1, 0)
+
+      // ensure all metadata are cleaned up
+      eventually(timeout(3.seconds), interval(200.milliseconds)) {
+        assert(metadataManager.getBatches(MetadataFilter(), 0, Int.MaxValue).isEmpty)
+      }
+
       // ensure no metadata request leak
       eventually(timeout(5.seconds), interval(200.milliseconds)) {
         assert(MetricsSystem.counterValue(METADATA_REQUEST_OPENED).getOrElse(0L) === 0)