[server] Integrate remote directory selector into table/partition creation

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -1619,6 +1619,43 @@ void testDynamicDiskWriteLimitRatio() throws Exception {
                 .get();
     }
 
+    @Test
+    void testDynamicRemoteDataDirsWithRoundRobin() throws Exception {
+        String originalDir = FLUSS_CLUSTER_EXTENSION.getRemoteDataDir();
+        String newDir1 = FLUSS_CLUSTER_EXTENSION.getRemoteDataDir("remote-dir-1");
+        String newDir2 = FLUSS_CLUSTER_EXTENSION.getRemoteDataDir("remote-dir-2");
+
+        // Dynamically switch from single remote.data.dir to multiple remote.data.dirs
+        // with round-robin strategy; original dir must be included
+        admin.alterClusterConfigs(
+                        Collections.singletonList(
+                                new AlterConfig(
+                                        ConfigOptions.REMOTE_DATA_DIRS.key(),
+                                        String.join(",", originalDir, newDir1, newDir2),
+                                        AlterConfigOpType.SET)))
+                .get();
+
+        // Create 6 tables and verify round-robin distribution across 3 directories
+        int tableCount = 6;
+        Map<String, Integer> dirUsageCount = new HashMap<>();
+        for (int i = 0; i < tableCount; i++) {
+            TablePath tablePath = TablePath.of("test_db", "dynamic_rr_table_" + i);
+            createTable(
+                    tablePath,
+                    TableDescriptor.builder().schema(DEFAULT_SCHEMA).distributedBy(1, "id").build(),
+                    true);
+
+            TableInfo tableInfo = admin.getTableInfo(tablePath).get();
+            String remoteDataDir = tableInfo.getRemoteDataDir();
+            assertThat(remoteDataDir).isNotNull();
+            dirUsageCount.merge(remoteDataDir, 1, Integer::sum);
+        }
+
+        // Each of the 3 directories should be used exactly twice (6 / 3 = 2)
+        assertThat(dirUsageCount).hasSize(3);
+        assertThat(dirUsageCount.values()).allMatch(count -> count == 2);
+    }
+
     private void assertConfigEntry(
             String key, @Nullable String value, ConfigEntry.ConfigSource source)
             throws ExecutionException, InterruptedException {

fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java

@@ -128,34 +128,9 @@ public static void validateTabletConfigs(Configuration conf) {
         validMinValue(ConfigOptions.TABLET_SERVER_ID, serverId.get(), 0);
     }
 
-    /** Validate common server configs. */
-    protected static void validateServerConfigs(Configuration conf) {
-        // Validate remote.data.dir and remote.data.dirs
-        String remoteDataDir = conf.get(ConfigOptions.REMOTE_DATA_DIR);
-        List<String> remoteDataDirs = conf.get(ConfigOptions.REMOTE_DATA_DIRS);
-        if (conf.get(ConfigOptions.REMOTE_DATA_DIR) == null
-                && conf.get(ConfigOptions.REMOTE_DATA_DIRS).isEmpty()) {
-            throw new IllegalConfigurationException(
-                    String.format(
-                            "Either %s or %s must be configured.",
-                            ConfigOptions.REMOTE_DATA_DIR.key(),
-                            ConfigOptions.REMOTE_DATA_DIRS.key()));
-        }
-
-        if (remoteDataDir != null) {
-            // Must validate that remote.data.dir is a valid FsPath
-            try {
-                new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR));
-            } catch (Exception e) {
-                throw new IllegalConfigurationException(
-                        String.format(
-                                "Invalid configuration for %s.",
-                                ConfigOptions.REMOTE_DATA_DIR.key()),
-                        e);
-            }
-        }
-
+    public static void validateRemoteDataDirs(Configuration conf) {
         // Validate remote.data.dirs
+        List<String> remoteDataDirs = conf.get(ConfigOptions.REMOTE_DATA_DIRS);
         for (int i = 0; i < remoteDataDirs.size(); i++) {
             String dir = remoteDataDirs.get(i);
             try {
@@ -185,19 +160,56 @@ protected static void validateServerConfigs(Configuration conf) {
                                     weights.size()));
                 }
 
-                // Validate all weights are no less than 0
+                // Verify that each weight is non-negative and that the total weight is greater than
+                // 0.
+                int totalWeight = 0;
                 for (int i = 0; i < weights.size(); i++) {
-                    if (weights.get(i) < 0) {
+                    int weight = weights.get(i);
+                    if (weight < 0) {
                         throw new IllegalConfigurationException(
                                 String.format(
                                         "All weights in '%s' must be no less than 0, but found %d at index %d.",
-                                        ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS.key(),
-                                        weights.get(i),
-                                        i));
+                                        ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS.key(), weight, i));
                     }
+                    totalWeight += weight;
+                }
+                if (totalWeight <= 0) {
+                    throw new IllegalConfigurationException(
+                            String.format(
+                                    "The sum of all weights in '%s' must be greater than 0, but the current sum is %d.",
+                                    ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS.key(), totalWeight));
                 }
             }
         }
+    }
+
+    /** Validate common server configs. */
+    protected static void validateServerConfigs(Configuration conf) {
+        // Validate remote.data.dir and remote.data.dirs
+        String remoteDataDir = conf.get(ConfigOptions.REMOTE_DATA_DIR);
+        List<String> remoteDataDirs = conf.get(ConfigOptions.REMOTE_DATA_DIRS);
+        if (remoteDataDir == null && remoteDataDirs.isEmpty()) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Either %s or %s must be configured.",
+                            ConfigOptions.REMOTE_DATA_DIR.key(),
+                            ConfigOptions.REMOTE_DATA_DIRS.key()));
+        }
+
+        if (remoteDataDir != null) {
+            // Must validate that remote.data.dir is a valid FsPath
+            try {
+                new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR));
+            } catch (Exception e) {
+                throw new IllegalConfigurationException(
+                        String.format(
+                                "Invalid configuration for %s.",
+                                ConfigOptions.REMOTE_DATA_DIR.key()),
+                        e);
+            }
+        }
+
+        validateRemoteDataDirs(conf);
 
         validMinValue(conf, ConfigOptions.DEFAULT_REPLICATION_FACTOR, 1);
         validMinValue(conf, ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, 1);

fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java

@@ -137,6 +137,20 @@ void testValidateCoordinatorConfigs() {
                 .hasMessageContaining(
                         "All weights in 'remote.data.dirs.weights' must be no less than 0");
 
+        // Test all zero weights
+        Configuration zeroWeightsConf = new Configuration();
+        zeroWeightsConf.set(
+                ConfigOptions.REMOTE_DATA_DIRS_STRATEGY,
+                ConfigOptions.RemoteDataDirStrategy.WEIGHTED_ROUND_ROBIN);
+        zeroWeightsConf.set(
+                ConfigOptions.REMOTE_DATA_DIRS, Arrays.asList("s3://bucket1", "s3://bucket2"));
+        zeroWeightsConf.set(ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS, Arrays.asList(0, 0));
+        assertThatThrownBy(() -> validateCoordinatorConfigs(zeroWeightsConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining("The sum of all weights")
+                .hasMessageContaining(ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS.key())
+                .hasMessageContaining("must be greater than 0");
+
         // Test invalid DEFAULT_REPLICATION_FACTOR
         Configuration invalidReplicationConf = new Configuration();
         invalidReplicationConf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");

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

@@ -45,6 +45,9 @@
 import static org.apache.fluss.config.ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC;
 import static org.apache.fluss.config.ConfigOptions.KV_SNAPSHOT_INTERVAL;
 import static org.apache.fluss.config.ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER;
+import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS;
+import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_STRATEGY;
+import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS;
 import static org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO;
 import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
 import static org.apache.fluss.utils.concurrent.LockUtils.inWriteLock;
@@ -66,7 +69,11 @@ class DynamicServerConfig {
                             LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(),
                             KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(),
                             KV_SNAPSHOT_INTERVAL.key(),
-                            SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key()));
+                            SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(),
+                            // Config options for remote.data.dirs
+                            REMOTE_DATA_DIRS.key(),
+                            REMOTE_DATA_DIRS_STRATEGY.key(),
+                            REMOTE_DATA_DIRS_WEIGHTS.key()));
     private static final Set<String> ALLOWED_CONFIG_PREFIXES = Collections.singleton("datalake.");
 
     private final ReadWriteLock lock = new ReentrantReadWriteLock();

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

@@ -302,6 +302,7 @@ public CompletableFuture<GetTableInfoResponse> getTableInfo(GetTableInfoRequest
         response.setTableJson(tableInfo.toTableDescriptor().toJsonBytes())
                 .setSchemaId(tableInfo.getSchemaId())
                 .setTableId(tableInfo.getTableId())
+                .setRemoteDataDir(tableInfo.getRemoteDataDir())
                 .setCreatedTime(tableInfo.getCreatedTime())
                 .setModifiedTime(tableInfo.getModifiedTime());
         return CompletableFuture.completedFuture(response);

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

@@ -29,6 +29,7 @@
 import org.apache.fluss.metadata.ResolvedPartitionSpec;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.coordinator.remote.RemoteDirDynamicLoader;
 import org.apache.fluss.server.metadata.ServerMetadataCache;
 import org.apache.fluss.server.zk.data.BucketAssignment;
 import org.apache.fluss.server.zk.data.PartitionAssignment;
@@ -85,6 +86,7 @@ public class AutoPartitionManager implements AutoCloseable {
 
     private final ServerMetadataCache metadataCache;
     private final MetadataManager metadataManager;
+    private final RemoteDirDynamicLoader remoteDirDynamicLoader;
     private final Clock clock;
 
     private final long periodicInterval;
@@ -108,10 +110,12 @@ public class AutoPartitionManager implements AutoCloseable {
     public AutoPartitionManager(
             ServerMetadataCache metadataCache,
             MetadataManager metadataManager,
+            RemoteDirDynamicLoader remoteDirDynamicLoader,
             Configuration conf) {
         this(
                 metadataCache,
                 metadataManager,
+                remoteDirDynamicLoader,
                 conf,
                 SystemClock.getInstance(),
                 Executors.newScheduledThreadPool(
@@ -122,11 +126,13 @@ public AutoPartitionManager(
     AutoPartitionManager(
             ServerMetadataCache metadataCache,
             MetadataManager metadataManager,
+            RemoteDirDynamicLoader remoteDirDynamicLoader,
             Configuration conf,
             Clock clock,
             ScheduledExecutorService periodicExecutor) {
         this.metadataCache = metadataCache;
         this.metadataManager = metadataManager;
+        this.remoteDirDynamicLoader = remoteDirDynamicLoader;
         this.clock = clock;
         this.periodicExecutor = periodicExecutor;
         this.periodicInterval = conf.get(ConfigOptions.AUTO_PARTITION_CHECK_INTERVAL).toMillis();
@@ -373,8 +379,10 @@ private void createPartitions(
                 PartitionAssignment partitionAssignment =
                         new PartitionAssignment(tableInfo.getTableId(), bucketAssignments);
 
+                // select a remote data dir for the partition
+                String remoteDataDir = remoteDirDynamicLoader.getRemoteDirSelector().nextDataDir();
                 metadataManager.createPartition(
-                        tablePath, tableId, partitionAssignment, partition, false);
+                        tablePath, tableId, remoteDataDir, partitionAssignment, partition, false);
                 // only single partition key table supports automatic creation of partitions
                 currentPartitions.put(partition.getPartitionName(), null);
                 LOG.info(

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

@@ -34,6 +34,7 @@
 import org.apache.fluss.server.authorizer.AuthorizerLoader;
 import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager;
 import org.apache.fluss.server.coordinator.rebalance.RebalanceManager;
+import org.apache.fluss.server.coordinator.remote.RemoteDirDynamicLoader;
 import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
 import org.apache.fluss.server.metadata.ServerMetadataCache;
 import org.apache.fluss.server.metrics.ServerMetricUtils;
@@ -148,6 +149,9 @@ public class CoordinatorServer extends ServerBase {
     @GuardedBy("lock")
     private LakeCatalogDynamicLoader lakeCatalogDynamicLoader;
 
+    @GuardedBy("lock")
+    private RemoteDirDynamicLoader remoteDirDynamicLoader;
+
     @GuardedBy("lock")
     private CoordinatorLeaderElection coordinatorLeaderElection;
 
@@ -225,10 +229,13 @@ protected void initCoordinatorStandby() throws Exception {
             this.coordinatorLeaderElection = new CoordinatorLeaderElection(zkClient, serverId);
 
             this.lakeCatalogDynamicLoader = new LakeCatalogDynamicLoader(conf, pluginManager, true);
+            this.remoteDirDynamicLoader = new RemoteDirDynamicLoader(conf);
+
             this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf, true);
 
             // Register server reconfigurable components
             dynamicConfigManager.register(lakeCatalogDynamicLoader);
+            dynamicConfigManager.register(remoteDirDynamicLoader);
 
             // Register stateless validators for coordinator-side upfront validation
             dynamicConfigManager.registerValidator(new DiskWriteLimitRatioValidator());
@@ -274,6 +281,7 @@ protected void initCoordinatorStandby() throws Exception {
                             authorizer,
                             lakeCatalogDynamicLoader,
                             lakeTableTieringManager,
+                            remoteDirDynamicLoader,
                             dynamicConfigManager,
                             ioExecutor,
                             kvSnapshotLeaseManager,
@@ -307,7 +315,8 @@ protected void initCoordinatorLeader() throws Exception {
             this.coordinatorChannelManager = new CoordinatorChannelManager(rpcClient);
 
             this.autoPartitionManager =
-                    new AutoPartitionManager(metadataCache, metadataManager, conf);
+                    new AutoPartitionManager(
+                            metadataCache, metadataManager, remoteDirDynamicLoader, conf);
             autoPartitionManager.start();
 
             // start coordinator event processor after we register coordinator leader to zk
@@ -620,6 +629,10 @@ CompletableFuture<Void> stopServices() {
                     lakeCatalogDynamicLoader.close();
                 }
 
+                if (remoteDirDynamicLoader != null) {
+                    remoteDirDynamicLoader.close();
+                }
+
                 if (kvSnapshotLeaseManager != null) {
                     kvSnapshotLeaseManager.close();
                 }

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

@@ -154,6 +154,7 @@
 import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager;
 import org.apache.fluss.server.coordinator.producer.ProducerOffsetsManager;
 import org.apache.fluss.server.coordinator.rebalance.goal.Goal;
+import org.apache.fluss.server.coordinator.remote.RemoteDirDynamicLoader;
 import org.apache.fluss.server.entity.CommitKvSnapshotData;
 import org.apache.fluss.server.entity.DatabasePropertyChanges;
 import org.apache.fluss.server.entity.LakeTieringTableInfo;
@@ -241,6 +242,7 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
     private final ProducerOffsetsManager producerOffsetsManager;
     private final KvSnapshotLeaseManager kvSnapshotLeaseManager;
     private final CoordinatorLeaderElection coordinatorLeaderElection;
+    private final RemoteDirDynamicLoader remoteDirDynamicLoader;
 
     public CoordinatorService(
             Configuration conf,
@@ -252,6 +254,7 @@ public CoordinatorService(
             @Nullable Authorizer authorizer,
             LakeCatalogDynamicLoader lakeCatalogDynamicLoader,
             LakeTableTieringManager lakeTableTieringManager,
+            RemoteDirDynamicLoader remoteDirDynamicLoader,
             DynamicConfigManager dynamicConfigManager,
             ExecutorService ioExecutor,
             KvSnapshotLeaseManager kvSnapshotLeaseManager,
@@ -278,6 +281,7 @@ public CoordinatorService(
         this.ioExecutor = ioExecutor;
         this.lakeTableHelper =
                 new LakeTableHelper(zkClient, conf.getString(ConfigOptions.REMOTE_DATA_DIR));
+        this.remoteDirDynamicLoader = remoteDirDynamicLoader;
 
         // Initialize and start the producer snapshot manager
         this.producerOffsetsManager = new ProducerOffsetsManager(conf, zkClient);
@@ -493,9 +497,18 @@ public CompletableFuture<CreateTableResponse> createTable(CreateTableRequest req
             }
         }
 
+        // select remote data dir for table.
+        // remote data dir will be used to store table data for non-partitioned table and metadata
+        // (such as lake snapshot offset file) for partitioned table
+        String remoteDataDir = remoteDirDynamicLoader.getRemoteDirSelector().nextDataDir();
+
         // then create table;
         metadataManager.createTable(
-                tablePath, tableDescriptor, tableAssignment, request.isIgnoreIfExists());
+                tablePath,
+                remoteDataDir,
+                tableDescriptor,
+                tableAssignment,
+                request.isIgnoreIfExists());
 
         return CompletableFuture.completedFuture(new CreateTableResponse());
     }
@@ -708,9 +721,13 @@ public CompletableFuture<CreatePartitionResponse> createPartition(
         PartitionAssignment partitionAssignment =
                 new PartitionAssignment(table.tableId, bucketAssignments);
 
+        // select remote data dir for partition
+        String remoteDataDir = remoteDirDynamicLoader.getRemoteDirSelector().nextDataDir();
+
         metadataManager.createPartition(
                 tablePath,
                 table.tableId,
+                remoteDataDir,
                 partitionAssignment,
                 partitionToCreate,
                 request.isIgnoreIfNotExists());
@@ -759,6 +776,7 @@ public CompletableFuture<MetadataResponse> metadata(MetadataRequest request) {
         return metadataResponseAccessContextEvent.getResultFuture();
     }
 
+    @Override
     public CompletableFuture<AdjustIsrResponse> adjustIsr(AdjustIsrRequest request) {
         CompletableFuture<AdjustIsrResponse> response = new CompletableFuture<>();
         eventManagerSupplier