[CELEBORN-1054] Support DB based ConfigService

common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala

@@ -368,6 +368,10 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
 
   def dynamicConfigStoreBackend: String = get(DYNAMIC_CONFIG_STORE_BACKEND)
   def dynamicConfigRefreshTime: Long = get(DYNAMIC_CONFIG_REFRESH_TIME)
+  def dynamicConfigStoreDbJdbcDriver: Option[String] = get(DYNAMIC_CONFIG_STORE_DB_JDBC_DRIVER)
+  def dynamicConfigStoreDbJdbcUrl: Option[String] = get(DYNAMIC_CONFIG_STORE_DB_JDBC_URL)
+  def dynamicConfigStoreDbJdbcUsername: Option[String] = get(DYNAMIC_CONFIG_STORE_DB_JDBC_USERNAME)
+  def dynamicConfigStoreDbJdbcPassword: Option[String] = get(DYNAMIC_CONFIG_STORE_DB_JDBC_PASSWORD)
 
   // //////////////////////////////////////////////////////
   //                      Network                        //
@@ -4166,17 +4170,50 @@ object CelebornConf extends Logging {
   val DYNAMIC_CONFIG_STORE_BACKEND: ConfigEntry[String] =
     buildConf("celeborn.dynamicConfig.store.backend")
       .categories("master", "worker")
-      .doc("Store backend for dynamic config, NONE means disabling dynamic config store")
+      .doc("Store backend for dynamic config. Available options: NONE, FS, DB. Note: NONE means disabling dynamic config store.")
       .version("0.4.0")
       .stringConf
-      .checkValues(Set("FS", "NONE"))
+      .transform(_.toUpperCase(Locale.ROOT))
+      .checkValues(Set("NONE", "FS", "DB"))
       .createWithDefault("NONE")
 
   val DYNAMIC_CONFIG_REFRESH_TIME: ConfigEntry[Long] =
     buildConf("celeborn.dynamicConfig.refresh.time")
       .categories("master", "worker")
       .version("0.4.0")
-      .doc("The time interval for refreshing the corresponding dynamic config periodically")
+      .doc("The time interval for refreshing the corresponding dynamic config periodically.")
       .timeConf(TimeUnit.MILLISECONDS)
       .createWithDefaultString("120s")
+
+  val DYNAMIC_CONFIG_STORE_DB_JDBC_DRIVER: OptionalConfigEntry[String] =
+    buildConf("celeborn.dynamicConfig.store.db.jdbc.driver")
+      .categories("master", "worker")
+      .version("0.4.0")
+      .doc("The jdbc driver of db store backend for dynamic config.")
+      .stringConf
+      .createOptional
+
+  val DYNAMIC_CONFIG_STORE_DB_JDBC_URL: OptionalConfigEntry[String] =
+    buildConf("celeborn.dynamicConfig.store.db.jdbc.url")
+      .categories("master", "worker")
+      .version("0.4.0")
+      .doc("The jdbc url of db store backend for dynamic config.")
+      .stringConf
+      .createOptional
+
+  val DYNAMIC_CONFIG_STORE_DB_JDBC_USERNAME: OptionalConfigEntry[String] =
+    buildConf("celeborn.dynamicConfig.store.db.jdbc.username")
+      .categories("master", "worker")
+      .version("0.4.0")
+      .doc("The jdbc username of db store backend for dynamic config.")
+      .stringConf
+      .createOptional
+
+  val DYNAMIC_CONFIG_STORE_DB_JDBC_PASSWORD: OptionalConfigEntry[String] =
+    buildConf("celeborn.dynamicConfig.db.jdbc.password")
+      .categories("master", "worker")
+      .version("0.4.0")
+      .doc("The jdbc password of db store backend for dynamic config.")
+      .stringConf
+      .createOptional
 }

docs/configuration/master.md

@@ -19,8 +19,12 @@ license: |
 <!--begin-include-->
 | Key | Default | Description | Since |
 | --- | ------- | ----------- | ----- |
-| celeborn.dynamicConfig.refresh.time | 120s | The time interval for refreshing the corresponding dynamic config periodically | 0.4.0 | 
-| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic config, NONE means disabling dynamic config store | 0.4.0 | 
+| celeborn.dynamicConfig.db.jdbc.password | &lt;undefined&gt; | The jdbc password of db store backend for dynamic config. | 0.4.0 | 
+| celeborn.dynamicConfig.refresh.time | 120s | The time interval for refreshing the corresponding dynamic config periodically. | 0.4.0 | 
+| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic config. Available options: NONE, FS, DB. Note: NONE means disabling dynamic config store. | 0.4.0 | 
+| celeborn.dynamicConfig.store.db.jdbc.driver | &lt;undefined&gt; | The jdbc driver of db store backend for dynamic config. | 0.4.0 | 
+| celeborn.dynamicConfig.store.db.jdbc.url | &lt;undefined&gt; | The jdbc url of db store backend for dynamic config. | 0.4.0 | 
+| celeborn.dynamicConfig.store.db.jdbc.username | &lt;undefined&gt; | The jdbc username of db store backend for dynamic config. | 0.4.0 | 
 | celeborn.master.estimatedPartitionSize.initialSize | 64mb | Initial partition size for estimation, it will change according to runtime stats. | 0.3.0 | 
 | celeborn.master.estimatedPartitionSize.update.initialDelay | 5min | Initial delay time before start updating partition size for estimation. | 0.3.0 | 
 | celeborn.master.estimatedPartitionSize.update.interval | 10min | Interval of updating partition size for estimation. | 0.3.0 | 

docs/configuration/worker.md

@@ -19,8 +19,12 @@ license: |
 <!--begin-include-->
 | Key | Default | Description | Since |
 | --- | ------- | ----------- | ----- |
-| celeborn.dynamicConfig.refresh.time | 120s | The time interval for refreshing the corresponding dynamic config periodically | 0.4.0 | 
-| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic config, NONE means disabling dynamic config store | 0.4.0 | 
+| celeborn.dynamicConfig.db.jdbc.password | &lt;undefined&gt; | The jdbc password of db store backend for dynamic config. | 0.4.0 | 
+| celeborn.dynamicConfig.refresh.time | 120s | The time interval for refreshing the corresponding dynamic config periodically. | 0.4.0 | 
+| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic config. Available options: NONE, FS, DB. Note: NONE means disabling dynamic config store. | 0.4.0 | 
+| celeborn.dynamicConfig.store.db.jdbc.driver | &lt;undefined&gt; | The jdbc driver of db store backend for dynamic config. | 0.4.0 | 
+| celeborn.dynamicConfig.store.db.jdbc.url | &lt;undefined&gt; | The jdbc url of db store backend for dynamic config. | 0.4.0 | 
+| celeborn.dynamicConfig.store.db.jdbc.username | &lt;undefined&gt; | The jdbc username of db store backend for dynamic config. | 0.4.0 | 
 | celeborn.master.endpoints | &lt;localhost&gt;:9097 | Endpoints of master nodes for celeborn client to connect, allowed pattern is: `<host1>:<port1>[,<host2>:<port2>]*`, e.g. `clb1:9097,clb2:9098,clb3:9099`. If the port is omitted, 9097 will be used. | 0.2.0 | 
 | celeborn.master.estimatedPartitionSize.minSize | 8mb | Ignore partition size smaller than this configuration of partition size for estimation. | 0.3.0 | 
 | celeborn.shuffle.chunk.size | 8m | Max chunk size of reducer's merged shuffle data. For example, if a reducer's shuffle data is 128M and the data will need 16 fetch chunk requests to fetch. | 0.2.0 | 

pom.xml

@@ -96,6 +96,7 @@
     <zstd-jni.version>1.5.2-1</zstd-jni.version>
     <kubernetes-client.version>6.7.0</kubernetes-client.version>
     <rocksdbjni.version>8.5.3</rocksdbjni.version>
+    <h2.version>2.2.224</h2.version>
 
     <shading.prefix>org.apache.celeborn.shaded</shading.prefix>
 
@@ -459,6 +460,12 @@
         <version>${kubernetes-client.version}</version>
         <scope>test</scope>
       </dependency>
+      <dependency>
+        <groupId>com.h2database</groupId>
+        <artifactId>h2</artifactId>
+        <version>${h2.version}</version>
+        <scope>test</scope>
+      </dependency>
       <dependency>
         <groupId>org.rocksdb</groupId>
         <artifactId>rocksdbjni</artifactId>

project/CelebornBuild.scala

@@ -61,6 +61,7 @@ object Dependencies {
   val scalatestVersion = "3.2.16"
   val slf4jVersion = "1.7.36"
   val snakeyamlVersion = "1.33"
+  val h2Version = "2.2.224"
 
   // Versions for proto
   val protocVersion = "3.19.2"
@@ -124,6 +125,7 @@ object Dependencies {
   val mockitoInline = "org.mockito" % "mockito-inline" % mockitoVersion
   val scalatestMockito = "org.mockito" %% "mockito-scala-scalatest" % scalatestMockitoVersion
   val scalatest = "org.scalatest" %% "scalatest" % scalatestVersion
+  val h2 = "com.h2database" % "h2" % h2Version
 }
 
 object CelebornCommonSettings {
@@ -223,7 +225,8 @@ object CelebornCommonSettings {
     Dependencies.scalatest % "test",
     Dependencies.junit % "test",
     // https://www.scala-sbt.org/1.x/docs/Testing.html
-    Dependencies.junitInterface % "test")
+    Dependencies.junitInterface % "test",
+    Dependencies.h2 % "test")
 }
 
 object CelebornBuild extends sbt.internal.BuildDef {

service/pom.xml

@@ -76,5 +76,10 @@
       <artifactId>log4j-1.2-api</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.h2database</groupId>
+      <artifactId>h2</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

service/src/main/java/org/apache/celeborn/server/common/service/config/BaseConfigServiceImpl.java

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.server.common.service.config;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import scala.concurrent.duration.Duration;
+
+import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.util.ThreadUtils;
+
+public abstract class BaseConfigServiceImpl implements ConfigService {
+  protected final CelebornConf celebornConf;
+  protected final AtomicReference<SystemConfig> systemConfigAtomicReference =
+      new AtomicReference<>();
+  protected final AtomicReference<Map<String, TenantConfig>> tenantConfigAtomicReference =
+      new AtomicReference<>(new HashMap<>());
+
+  private final ScheduledExecutorService configRefreshService =
+      ThreadUtils.newDaemonSingleThreadScheduledExecutor("config-refresh-service");
+
+  protected static final String CONF_TENANT_ID = "tenantId";
+  protected static final String CONF_LEVEL = "level";
+  protected static final String CONF_CONFIG = "config";
+
+  public BaseConfigServiceImpl(CelebornConf celebornConf) {
+    this.celebornConf = celebornConf;
+    this.init();
+    this.refresh();
+    long dynamicConfigRefreshTime = celebornConf.dynamicConfigRefreshTime();
+    this.configRefreshService.scheduleWithFixedDelay(
+        this::refresh, dynamicConfigRefreshTime, dynamicConfigRefreshTime, TimeUnit.MILLISECONDS);
+  }
+
+  protected abstract void init();
+
+  protected abstract void refresh();
+
+  protected void refreshConfig(List<Map<String, Object>> dynamicConfigs) {
+    if (dynamicConfigs == null) {
+      return;
+    }
+    SystemConfig systemConfig = null;
+    Map<String, TenantConfig> tenantConfigs = new HashMap<>();
+    for (Map<String, Object> settings : dynamicConfigs) {
+      String tenantId = (String) settings.get(CONF_TENANT_ID);
+      String level = (String) settings.get(CONF_LEVEL);
+      Map<String, String> config =
+          ((Map<String, Object>) settings.get(CONF_CONFIG))
+              .entrySet().stream()
+                  .collect(Collectors.toMap(Map.Entry::getKey, a -> a.getValue().toString()));
+      if (ConfigLevel.SYSTEM.name().equalsIgnoreCase(level)) {
+        systemConfig = new SystemConfig(celebornConf, config);
+      } else if (ConfigLevel.TENANT.name().equalsIgnoreCase(level)) {
+        TenantConfig tenantConfig = new TenantConfig(this, tenantId, config);
+        tenantConfigs.put(tenantId, tenantConfig);
+      }
+    }
+    systemConfigAtomicReference.set(
+        systemConfig == null ? new SystemConfig(celebornConf) : systemConfig);
+    tenantConfigAtomicReference.set(tenantConfigs);
+  }
+
+  @Override
+  public SystemConfig getSystemConfig() {
+    return systemConfigAtomicReference.get();
+  }
+
+  @Override
+  public TenantConfig getRawTenantConfig(String tenantId) {
+    return tenantConfigAtomicReference.get().get(tenantId);
+  }
+
+  @Override
+  public void refreshAllCache() {
+    this.refresh();
+  }
+
+  @Override
+  public void close() {
+    ThreadUtils.shutdown(configRefreshService, Duration.apply("800ms"));
+  }
+}

service/src/main/scala/org/apache/celeborn/server/common/service/config/ConfigService.java → service/src/main/java/org/apache/celeborn/server/common/service/config/ConfigService.java

@@ -17,7 +17,7 @@
 
 package org.apache.celeborn.server.common.service.config;
 
-public interface ConfigService {
+public interface ConfigService extends AutoCloseable {
 
   SystemConfig getSystemConfig();
 
@@ -33,7 +33,4 @@ default DynamicConfig getTenantConfig(String tenantId) {
   }
 
   void refreshAllCache();
-
-  void shutdown();
-
-}
+}

service/src/main/java/org/apache/celeborn/server/common/service/config/ConfigStoreBackend.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.server.common.service.config;
+
+public enum ConfigStoreBackend {
+  NONE,
+  FS,
+  DB;
+}