modified based on CR2

loserwang1024 · loserwang1024 · commit 6ec7ee04f1cf · 2025-10-10T20:35:04.000+08:00
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
@@ -21,10 +21,10 @@
 import org.apache.fluss.client.metadata.KvSnapshotMetadata;
 import org.apache.fluss.client.metadata.KvSnapshots;
 import org.apache.fluss.client.metadata.LakeSnapshot;
-import org.apache.fluss.cluster.AlterConfig;
-import org.apache.fluss.cluster.ConfigEntry;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.exception.DatabaseAlreadyExistException;
 import org.apache.fluss.exception.DatabaseNotEmptyException;
 import org.apache.fluss.exception.DatabaseNotExistException;
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
@@ -22,10 +22,10 @@
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.utils.ClientRpcMessageUtils;
-import org.apache.fluss.cluster.AlterConfig;
 import org.apache.fluss.cluster.Cluster;
-import org.apache.fluss.cluster.ConfigEntry;
 import org.apache.fluss.cluster.ServerNode;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.exception.LeaderNotAvailableException;
 import org.apache.fluss.metadata.DatabaseDescriptor;
 import org.apache.fluss.metadata.DatabaseInfo;
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
@@ -25,7 +25,7 @@
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.client.write.KvWriteBatch;
 import org.apache.fluss.client.write.ReadyWriteBatch;
-import org.apache.fluss.cluster.ConfigEntry;
+import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.fs.FsPathAndFileName;
 import org.apache.fluss.fs.token.ObtainedSecurityToken;
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -23,12 +23,12 @@
 import org.apache.fluss.client.metadata.KvSnapshots;
 import org.apache.fluss.client.table.Table;
 import org.apache.fluss.client.table.writer.UpsertWriter;
-import org.apache.fluss.cluster.AlterConfig;
-import org.apache.fluss.cluster.ConfigEntry;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.config.AutoPartitionTimeUnit;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.exception.DatabaseAlreadyExistException;
 import org.apache.fluss.exception.DatabaseNotEmptyException;
 import org.apache.fluss.exception.DatabaseNotExistException;
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
@@ -25,11 +25,11 @@
 import org.apache.fluss.client.table.scanner.batch.BatchScanner;
 import org.apache.fluss.client.table.writer.AppendWriter;
 import org.apache.fluss.client.utils.ClientRpcMessageUtils;
-import org.apache.fluss.cluster.AlterConfig;
-import org.apache.fluss.cluster.ConfigEntry;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.exception.AuthorizationException;
 import org.apache.fluss.metadata.AlterConfigOpType;
 import org.apache.fluss.metadata.DataLakeFormat;
@@ -719,6 +719,17 @@ void testDynamicConfigs() throws ExecutionException, InterruptedException {
                                 new AlterConfig(
                                         DATALAKE_FORMAT.key(), null, AlterConfigOpType.SET)))
                 .get();
+        assertThat(guestAdmin.describeClusterConfigs().get())
+                .contains(
+                        new ConfigEntry(
+                                DATALAKE_FORMAT.key(),
+                                null,
+                                ConfigEntry.ConfigSource.DYNAMIC_SERVER_CONFIG))
+                .doesNotContain(
+                        new ConfigEntry(
+                                DATALAKE_FORMAT.key(),
+                                "paimon",
+                                ConfigEntry.ConfigSource.INITIAL_SERVER_CONFIG));
     }
 
     private static Configuration initConfig() {
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/cluster/AlterConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/cluster/AlterConfig.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.cluster;
+package org.apache.fluss.config.cluster;
 
 import org.apache.fluss.metadata.AlterConfigOpType;
 
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/cluster/ConfigEntry.java b/fluss-common/src/main/java/org/apache/fluss/config/cluster/ConfigEntry.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.cluster;
+package org.apache.fluss.config.cluster;
 
 import javax.annotation.Nullable;
 
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/cluster/ServerReconfigurable.java b/fluss-common/src/main/java/org/apache/fluss/config/cluster/ServerReconfigurable.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.cluster;
+package org.apache.fluss.config.cluster;
 
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.ConfigException;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
@@ -18,10 +18,10 @@
 package org.apache.fluss.server;
 
 import org.apache.fluss.annotation.VisibleForTesting;
-import org.apache.fluss.cluster.AlterConfig;
-import org.apache.fluss.cluster.ConfigEntry;
-import org.apache.fluss.cluster.ServerReconfigurable;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.ConfigEntry;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
 import org.apache.fluss.exception.ConfigException;
 import org.apache.fluss.server.authorizer.ZkNodeChangeNotificationWatcher;
 import org.apache.fluss.server.zk.ZooKeeperClient;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
@@ -18,8 +18,8 @@
 package org.apache.fluss.server;
 
 import org.apache.fluss.annotation.Internal;
-import org.apache.fluss.cluster.ServerReconfigurable;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
 import org.apache.fluss.exception.ConfigException;
 import org.apache.fluss.utils.MapUtils;
 
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
@@ -17,9 +17,9 @@
 
 package org.apache.fluss.server;
 
-import org.apache.fluss.cluster.ConfigEntry;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.exception.KvSnapshotNotExistException;
 import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -18,11 +18,11 @@
 package org.apache.fluss.server.coordinator;
 
 import org.apache.fluss.annotation.VisibleForTesting;
-import org.apache.fluss.cluster.AlterConfig;
 import org.apache.fluss.cluster.ServerType;
 import org.apache.fluss.cluster.TabletServerInfo;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.AlterConfig;
 import org.apache.fluss.exception.InvalidAlterTableException;
 import org.apache.fluss.exception.InvalidCoordinatorException;
 import org.apache.fluss.exception.InvalidDatabaseException;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeCatalogDynamicLoader.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeCatalogDynamicLoader.java
@@ -17,9 +17,9 @@
 
 package org.apache.fluss.server.coordinator;
 
-import org.apache.fluss.cluster.ServerReconfigurable;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.ServerReconfigurable;
 import org.apache.fluss.exception.ConfigException;
 import org.apache.fluss.lake.lakestorage.LakeCatalog;
 import org.apache.fluss.lake.lakestorage.LakeStorage;
@@ -65,7 +65,6 @@ public void validate(Configuration newConfig) throws ConfigException {
                         : currentConfiguration.get(DATALAKE_FORMAT);
         Map<String, String> configMap = newConfig.toMap();
         String datalakePrefix = "datalake." + newDatalakeFormat + ".";
-        final DataLakeFormat finalDatalakeFormat = newDatalakeFormat;
         configMap.forEach(
                 (key, value) -> {
                     if (!key.equals(DATALAKE_FORMAT.key())
@@ -74,7 +73,7 @@ public void validate(Configuration newConfig) throws ConfigException {
                         throw new ConfigException(
                                 String.format(
                                         "Invalid configuration %s for %s datalake format",
-                                        newConfig, finalDatalakeFormat));
+                                        key, newDatalakeFormat));
                     }
                 });
     }
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -17,11 +17,11 @@
 
 package org.apache.fluss.server.utils;
 
-import org.apache.fluss.cluster.ConfigEntry;
 import org.apache.fluss.cluster.Endpoint;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.ServerType;
 import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.cluster.ConfigEntry;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.fs.token.ObtainedSecurityToken;
 import org.apache.fluss.metadata.AlterConfigOpType;
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
@@ -17,9 +17,9 @@
 
 package org.apache.fluss.server;
 
-import org.apache.fluss.cluster.AlterConfig;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.AlterConfig;
 import org.apache.fluss.exception.ConfigException;
 import org.apache.fluss.metadata.AlterConfigOpType;
 import org.apache.fluss.server.coordinator.LakeCatalogDynamicLoader;
@@ -202,7 +202,7 @@ void testWrongLakeFormatPrefix() throws Exception {
                                                             "filesystem",
                                                             AlterConfigOpType.SET))))
                     .hasMessage(
-                            "Invalid configuration {datalake.format=paimon, datalake.iceberg.metastore=filesystem} for paimon datalake format");
+                            "Invalid configuration datalake.iceberg.metastore for paimon datalake format");
 
             assertThat(lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat())
                     .isNull();
