Skip to content

Commit f6797c9

Browse files
turboFeiturboFei
committed
Save
1 parent c94d254 commit f6797c9

File tree

9 files changed

+81
-24
lines changed

9 files changed

+81
-24
lines changed

‎docs/configuration/settings.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -382,6 +382,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
382382

383383
| Key | Default | Meaning | Type | Since |
384384
|-------------------------------------------------|----------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------|
385+
| kyuubi.metadata.cleaner.batch.size | 2147483647 | The batch size for cleaning expired metadata. This is used to avoid holding the delete lock for a long time when there are too many expired metadata to be cleaned. | int | 1.11.0 |
385386
| kyuubi.metadata.cleaner.enabled | true | Whether to clean the metadata periodically. If it is enabled, Kyuubi will clean the metadata that is in the terminate state with max age limitation. | boolean | 1.6.0 |
386387
| kyuubi.metadata.cleaner.interval | PT30M | The interval to check and clean expired metadata. | duration | 1.6.0 |
387388
| kyuubi.metadata.max.age | PT72H | The maximum age of metadata, the metadata exceeding the age will be cleaned. | duration | 1.6.0 |

‎integration-tests/kyuubi-jdbc-it/src/test/scala/org/apache/kyuubi/server/metadata/MySQLMetadataManagerSuite.scala

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.kyuubi.server.metadata
19+
20+
import org.apache.kyuubi.server.metadata.jdbc.JDBCMetadataStoreConf
21+
import org.apache.kyuubi.server.mysql.MySQLJDBCTestHelper
22+
23+
class MySQLMetadataManagerSuite extends MetadataManagerSuite with MySQLJDBCTestHelper {
24+
25+
override def beforeAll(): Unit = {
26+
kyuubiConf.set(JDBCMetadataStoreConf.METADATA_STORE_JDBC_DATABASE_TYPE.key, "MYSQL")
27+
kyuubiConf.set(JDBCMetadataStoreConf.METADATA_STORE_JDBC_URL.key, jdbcUrlWithConf("jdbc:mysql://localhost:3306/kyuubi_metadata"))
28+
29+
}
30+
31+
32+
}

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

Lines changed: 11 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -2077,25 +2077,24 @@ object KyuubiConf {
20772077
.timeConf
20782078
.createWithDefault(Duration.ofMinutes(30).toMillis)
20792079

2080-
val METADATA_CLEANER_BATCH_INTERVAL: ConfigEntry[Long] =
2081-
buildConf("kyuubi.metadata.cleaner.batch.interval")
2082-
.serverOnly
2083-
.internal
2084-
.doc("The interval to check and clean expired metadata in batch mode. " +
2085-
"This is used to avoid the metadata cleaner thread being blocked for a long time " +
2086-
"when there are too many expired metadata to be cleaned.")
2087-
.version("1.11.0")
2088-
.timeConf
2089-
.createWithDefault(Duration.ofMillis(100).toMillis)
2090-
20912080
val METADATA_CLEANER_BATCH_SIZE: ConfigEntry[Int] =
20922081
buildConf("kyuubi.metadata.cleaner.batch.size")
20932082
.serverOnly
2094-
.doc("The batch size for cleaning expired metadata.")
2083+
.doc("The batch size for cleaning expired metadata. " +
2084+
"This is used to avoid holding the delete lock for a long time " +
2085+
"when there are too many expired metadata to be cleaned.")
20952086
.version("1.11.0")
20962087
.intConf
20972088
.createWithDefault(Int.MaxValue)
20982089

2090+
val METADATA_CLEANER_BATCH_INTERVAL: ConfigEntry[Long] =
2091+
buildConf("kyuubi.metadata.cleaner.batch.interval")
2092+
.serverOnly
2093+
.internal
2094+
.doc("The interval to wait before next batch of cleaning expired metadata.")
2095+
.timeConf
2096+
.createWithDefault(Duration.ofMillis(100).toMillis)
2097+
20992098
val METADATA_RECOVERY_THREADS: ConfigEntry[Int] =
21002099
buildConf("kyuubi.metadata.recovery.threads")
21012100
.serverOnly

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,13 @@
1717

1818
package org.apache.kyuubi.server.metadata
1919

20-
import com.google.common.annotations.VisibleForTesting
21-
2220
import java.util.concurrent.{ConcurrentHashMap, ThreadPoolExecutor, TimeUnit}
2321
import java.util.concurrent.atomic.AtomicInteger
22+
2423
import scala.collection.JavaConverters._
24+
25+
import com.google.common.annotations.VisibleForTesting
26+
2527
import org.apache.kyuubi.{KyuubiException, Logging}
2628
import org.apache.kyuubi.client.api.v1.dto.Batch
2729
import org.apache.kyuubi.config.KyuubiConf

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,10 @@ trait MetadataStore extends Closeable {
106106
*/
107107
def cleanupMetadataByAge(maxAge: Long, limit: Int): Int
108108

109+
def cleanupMetadataByAge(maxAge: Long): Int = {
110+
cleanupMetadataByAge(maxAge, Int.MaxValue)
111+
}
112+
109113
/**
110114
* Cleanup kubernetes engine info by identifier.
111115
*/
@@ -117,4 +121,8 @@ trait MetadataStore extends Closeable {
117121
* @param limit the maximum number of kubernetes engine info to be cleaned up.
118122
*/
119123
def cleanupKubernetesEngineInfoByAge(maxAge: Long, limit: Int): Int
124+
125+
def cleanupKubernetesEngineInfoByAge(maxAge: Long): Int = {
126+
cleanupKubernetesEngineInfoByAge(maxAge, Int.MaxValue)
127+
}
120128
}

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

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -414,9 +414,10 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
414414
override def cleanupMetadataByAge(maxAge: Long, limit: Int): Int = {
415415
val minEndTime = System.currentTimeMillis() - maxAge
416416
val query =
417-
s"DELETE FROM $METADATA_TABLE WHERE state IN ($terminalStates) AND end_time < ? liMIT ?"
417+
s"DELETE FROM $METADATA_TABLE WHERE state IN ($terminalStates) AND end_time < ?" +
418+
s" ${dialect.deleteFromLimitClause(limit)}"
418419
JdbcUtils.withConnection { connection =>
419-
withUpdateCount(connection, query, minEndTime, limit) { count =>
420+
withUpdateCount(connection, query, minEndTime) { count =>
420421
info(s"Cleaned up $count records older than $maxAge ms from $METADATA_TABLE limit:$limit.")
421422
count
422423
}
@@ -467,9 +468,10 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
467468

468469
override def cleanupKubernetesEngineInfoByAge(maxAge: Long, limit: Int): Int = {
469470
val minUpdateTime = System.currentTimeMillis() - maxAge
470-
val query = s"DELETE FROM $KUBERNETES_ENGINE_INFO_TABLE WHERE update_time < ? LIMIT ?"
471+
val query = s"DELETE FROM $KUBERNETES_ENGINE_INFO_TABLE WHERE update_time < ?" +
472+
s" ${dialect.deleteFromLimitClause(limit)}"
471473
JdbcUtils.withConnection { connection =>
472-
withUpdateCount(connection, query, minUpdateTime, limit) { count =>
474+
withUpdateCount(connection, query, minUpdateTime) { count =>
473475
info(s"Cleaned up $count records older than $maxAge ms from $KUBERNETES_ENGINE_INFO_TABLE" +
474476
s" limit $limit.")
475477
count

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package org.apache.kyuubi.server.metadata.jdbc
1919

2020
trait JdbcDatabaseDialect {
2121
def limitClause(limit: Int, offset: Int): String
22+
def deleteFromLimitClause(limit: Int): String
2223
def insertOrReplace(
2324
table: String,
2425
cols: Seq[String],
@@ -31,6 +32,10 @@ class GenericDatabaseDialect extends JdbcDatabaseDialect {
3132
s"LIMIT $limit OFFSET $offset"
3233
}
3334

35+
override def deleteFromLimitClause(limit: Int): String = {
36+
"" // Generic dialect does not support LIMIT in DELETE statements
37+
}
38+
3439
override def insertOrReplace(
3540
table: String,
3641
cols: Seq[String],
@@ -71,6 +76,10 @@ class MySQLDatabaseDialect extends GenericDatabaseDialect {
7176
|${cols.filterNot(_ == keyCol).map(c => s"$c = new.$c").mkString(",")}
7277
|""".stripMargin
7378
}
79+
80+
override def deleteFromLimitClause(limit: Int): String = {
81+
s"LIMIT $limit"
82+
}
7483
}
7584
class PostgreSQLDatabaseDialect extends GenericDatabaseDialect {
7685
override def insertOrReplace(

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

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ import org.apache.kyuubi.server.metadata.jdbc.JDBCMetadataStoreConf.METADATA_STO
3434
import org.apache.kyuubi.session.SessionType
3535

3636
class MetadataManagerSuite extends KyuubiFunSuite {
37+
protected val kyuubiConf = new KyuubiConf()
3738

3839
test("fail fast on duplicated key") {
3940
Seq("true", "false").foreach { enableAsyncRetry =>
@@ -203,6 +204,7 @@ class MetadataManagerSuite extends KyuubiFunSuite {
203204
val metricsSystem = new MetricsSystem()
204205
val metadataManager = newMetadataMgr()
205206
val conf = KyuubiConf()
207+
kyuubiConf.getAll.foreach { case (k, v) => conf.set(k, v) }
206208
confOverlay.foreach { case (k, v) => conf.set(k, v) }
207209
try {
208210
metricsSystem.initialize(conf)
@@ -214,10 +216,12 @@ class MetadataManagerSuite extends KyuubiFunSuite {
214216
metadataManager.getBatches(MetadataFilter(), 0, Int.MaxValue).foreach { batch =>
215217
// close the batch if not ended
216218
if (batch.getEndTime == 0) {
217-
metadataManager.updateMetadata(Metadata(
218-
identifier = batch.getId,
219-
state = OperationState.CLOSED.toString,
220-
endTime = System.currentTimeMillis()), false)
219+
metadataManager.updateMetadata(
220+
Metadata(
221+
identifier = batch.getId,
222+
state = OperationState.CLOSED.toString,
223+
endTime = System.currentTimeMillis()),
224+
false)
221225
}
222226
}
223227

‎kyuubi-server/src/test/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStoreSuite.scala

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ class JDBCMetadataStoreSuite extends KyuubiFunSuite {
4545
batch =>
4646
jdbcMetadataStore.cleanupMetadataByIdentifier(batch.identifier)
4747
}
48-
jdbcMetadataStore.cleanupKubernetesEngineInfoByAge(0, Int.MaxValue)
48+
jdbcMetadataStore.cleanupKubernetesEngineInfoByAge(0)
4949
jdbcMetadataStore.close()
5050
}
5151

@@ -242,7 +242,7 @@ class JDBCMetadataStoreSuite extends KyuubiFunSuite {
242242
Int.MaxValue).isEmpty)
243243

244244
eventually(Timeout(3.seconds)) {
245-
jdbcMetadataStore.cleanupMetadataByAge(1000, Int.MaxValue)
245+
jdbcMetadataStore.cleanupMetadataByAge(1000)
246246
assert(jdbcMetadataStore.getMetadata(batchId) == null)
247247
}
248248
}

0 commit comments

Comments
 (0)