[Enhancement](FE) Make scheduler aware of interval changes immediately

fe/fe-common/src/main/java/org/apache/doris/common/ConfigBase.java

@@ -123,6 +123,10 @@ public void init(String configFile) throws Exception {
         }
     }
 
+    public static Field getField(String name) {
+        return confFields.get(name);
+    }
+
     public void initCustom(String customConfFile) throws Exception {
         this.customConfFile = customConfFile;
         File file = new File(customConfFile);

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -333,6 +333,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 
@@ -566,6 +567,10 @@ public class Env {
 
     private final List<String> forceSkipJournalIds = Arrays.asList(Config.force_skip_journal_ids);
 
+    // if a config is relative to a daemon thread. record the relation here. we will proactively change interval of it.
+    private final Map<String, Supplier<MasterDaemon>> configtoThreads = ImmutableMap
+            .of("dynamic_partition_check_interval_seconds", this::getDynamicPartitionScheduler);
+
     public List<TFrontendInfo> getFrontendInfos() {
         List<TFrontendInfo> res = new ArrayList<>();
 
@@ -5705,13 +5710,30 @@ public void replayDropGlobalFunction(FunctionSearchDesc functionSearchDesc) {
         globalFunctionMgr.replayDropFunction(functionSearchDesc);
     }
 
+    /**
+     * we can't set callback which is in fe-core to config items which are in fe-common. so wrap them here. it's not so
+     * good but is best for us now.
+     */
+    public void setMutableConfigwithCallback(String key, String value) throws ConfigException {
+        ConfigBase.setMutableConfig(key, value);
+        if (configtoThreads.get(key) != null) {
+            try {
+                configtoThreads.get(key).get().setInterval(Config.getField(key).getLong(null) * 1000L);
+                configtoThreads.get(key).get().interrupt();
+                LOG.info("set config " + key + " to " + value);
+            } catch (IllegalAccessException e) {
+                LOG.warn("set config " + key + " failed: " + e.getMessage());
+            }
+        }
+    }
+
     public void setConfig(AdminSetConfigStmt stmt) throws Exception {
         Map<String, String> configs = stmt.getConfigs();
         Preconditions.checkState(configs.size() == 1);
 
         for (Map.Entry<String, String> entry : configs.entrySet()) {
             try {
-                ConfigBase.setMutableConfig(entry.getKey(), entry.getValue());
+                setMutableConfigwithCallback(entry.getKey(), entry.getValue());
             } catch (ConfigException e) {
                 throw new DdlException(e.getMessage());
             }

fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/SetConfigAction.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.httpv2.rest;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.ConfigBase;
 import org.apache.doris.common.ConfigException;
 import org.apache.doris.common.DdlException;
@@ -93,7 +94,7 @@ protected Object set_config(HttpServletRequest request, HttpServletResponse resp
             try {
                 if (confValue != null && confValue.length == 1) {
                     try {
-                        ConfigBase.setMutableConfig(confKey, confValue[0]);
+                        Env.getCurrentEnv().setMutableConfigwithCallback(confKey, confValue[0]);
                     } catch (ConfigException e) {
                         throw new DdlException(e.getMessage());
                     }