[server] Skip datalake prefix validation when datalake format is not null

fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java

@@ -210,6 +210,7 @@ private void processDeletions(
             throws ConfigException {
         for (String configKey : dynamicConfigs.keySet()) {
             if (newDynamicConfigs.containsKey(configKey)) {
+                effectiveChanges.put(configKey, newDynamicConfigs.get(configKey));
                 continue; // Not deleted
             }
 

fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeCatalogDynamicLoader.java

@@ -63,6 +63,12 @@ public void validate(Configuration newConfig) throws ConfigException {
                 newConfig.getOptional(DATALAKE_FORMAT).isPresent()
                         ? newConfig.get(DATALAKE_FORMAT)
                         : currentConfiguration.get(DATALAKE_FORMAT);
+        // If datalake format is not set, skip prefix validation so that users can disable or enable
+        // datalake format without re-supplying all datalake-prefixed properties.
+        if (newDatalakeFormat == null) {
+            return;
+        }
+
         Map<String, String> configMap = newConfig.toMap();
         String datalakePrefix = "datalake." + newDatalakeFormat + ".";
         configMap.forEach(

fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java

@@ -184,6 +184,47 @@ void testUnknownLakeHouse() throws Exception {
         }
     }
 
+    @Test
+    void testDatalakePrefixValidationSkippedWhenFormatIsNull() throws Exception {
+        Configuration configuration = new Configuration();
+        try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
+                new LakeCatalogDynamicLoader(configuration, null, true)) {
+            DynamicConfigManager dynamicConfigManager =
+                    new DynamicConfigManager(zookeeperClient, configuration, true);
+            dynamicConfigManager.register(lakeCatalogDynamicLoader);
+            dynamicConfigManager.startup();
+
+            // Setting `datalake.paimon.*` without setting `datalake.format` should pass because
+            // prefix validation is skipped.
+            assertThatCode(
+                            () ->
+                                    dynamicConfigManager.alterConfigs(
+                                            Collections.singletonList(
+                                                    new AlterConfig(
+                                                            "datalake.iceberg.type",
+                                                            "rest",
+                                                            AlterConfigOpType.SET))))
+                    .doesNotThrowAnyException();
+
+            assertThat(lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat())
+                    .isNull();
+            assertThatThrownBy(
+                            () ->
+                                    dynamicConfigManager.alterConfigs(
+                                            Arrays.asList(
+                                                    new AlterConfig(
+                                                            "datalake.iceberg.type",
+                                                            "rest",
+                                                            AlterConfigOpType.SET),
+                                                    new AlterConfig(
+                                                            "datalake.format",
+                                                            "paimon",
+                                                            AlterConfigOpType.SET))))
+                    .hasMessageContaining(
+                            "Invalid configuration 'datalake.iceberg.type' for 'paimon' datalake format");
+        }
+    }
+
     @Test
     void testWrongLakeFormatPrefix() throws Exception {
         Configuration configuration = new Configuration();