[server] Support Multi-Location for Remote Storage

Author: LiebingYu

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java

@@ -167,8 +167,9 @@ void fetchOnce() throws Exception {
                             (bytes, throwable) -> {
                                 if (throwable != null) {
                                     LOG.error(
-                                            "Failed to download remote log segment file {}.",
+                                            "Failed to download remote log segment file {} for bucket {}.",
                                             fsPathAndFileName.getFileName(),
+                                            request.segment.tableBucket(),
                                             ExceptionUtils.stripExecutionException(throwable));
                                     // release the semaphore for the failed request
                                     prefetchSemaphore.release();
@@ -178,8 +179,9 @@ void fetchOnce() throws Exception {
                                     scannerMetricGroup.remoteFetchErrorCount().inc();
                                 } else {
                                     LOG.info(
-                                            "Successfully downloaded remote log segment file {} to local cost {} ms.",
+                                            "Successfully downloaded remote log segment file {} for bucket {} to local cost {} ms.",
                                             fsPathAndFileName.getFileName(),
+                                            request.segment.tableBucket(),
                                             System.currentTimeMillis() - startTime);
                                     File localFile =
                                             new File(

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

@@ -96,6 +96,44 @@ public class ConfigOptions {
                             "The directory used for storing the kv snapshot data files and remote log for log tiered storage "
                                     + " in a Fluss supported filesystem.");
 
+    public static final ConfigOption<List<String>> REMOTE_DATA_DIRS =
+            key("remote.data.dirs")
+                    .stringType()
+                    .asList()
+                    .defaultValues()
+                    .withDescription(
+                            "The directories used for storing the kv snapshot data files and remote log for log tiered storage "
+                                    + " in a Fluss supported filesystem. "
+                                    + "This is a list of remote data directory paths. "
+                                    + "Example: `remote.data.dirs: oss://bucket1/fluss-remote-data, oss://bucket2/fluss-remote-data`.");
+
+    public static final ConfigOption<RemoteDataDirStrategy> REMOTE_DATA_DIRS_STRATEGY =
+            key("remote.data.dirs.strategy")
+                    .enumType(RemoteDataDirStrategy.class)
+                    .defaultValue(RemoteDataDirStrategy.ROUND_ROBIN)
+                    .withDescription(
+                            "The strategy for selecting the remote data directory from `"
+                                    + REMOTE_DATA_DIRS.key()
+                                    + "`.");
+
+    public static final ConfigOption<List<Integer>> REMOTE_DATA_DIRS_WEIGHTS =
+            key("remote.data.dirs.weights")
+                    .intType()
+                    .asList()
+                    .defaultValues()
+                    .withDescription(
+                            "The weights of the remote data directories. "
+                                    + "This is a list of weights corresponding to the `"
+                                    + REMOTE_DATA_DIRS.key()
+                                    + "` in the same order. When `"
+                                    + REMOTE_DATA_DIRS_STRATEGY.key()
+                                    + "` is set to `"
+                                    + RemoteDataDirStrategy.WEIGHTED_ROUND_ROBIN
+                                    + "`, this must be configured, and its size must be equal to `"
+                                    + REMOTE_DATA_DIRS.key()
+                                    + "`; otherwise, it will be ignored."
+                                    + "Example: `remote.data.dir.weights: 1, 2`");
+
     public static final ConfigOption<MemorySize> REMOTE_FS_WRITE_BUFFER_SIZE =
             key("remote.fs.write-buffer-size")
                     .memoryType()
@@ -1925,4 +1963,10 @@ private static class ConfigOptionsHolder {
     public static ConfigOption<?> getConfigOption(String key) {
         return ConfigOptionsHolder.CONFIG_OPTIONS_BY_KEY.get(key);
     }
+
+    /** Remote data dir select strategy for Fluss. */
+    public enum RemoteDataDirStrategy {
+        ROUND_ROBIN,
+        WEIGHTED_ROUND_ROBIN
+    }
 }

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

@@ -19,12 +19,14 @@
 
 import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.exception.IllegalConfigurationException;
 
 import java.lang.reflect.Field;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 /** Utilities of Fluss {@link ConfigOptions}. */
 @Internal
@@ -74,4 +76,97 @@ static Map<String, ConfigOption<?>> extractConfigOptions(String prefix) {
         }
         return options;
     }
+
+    public static void validateCoordinatorConfigs(Configuration conf) {
+        validServerConfigs(conf);
+
+        if (conf.get(ConfigOptions.DEFAULT_REPLICATION_FACTOR) < 1) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be greater than or equal 1.",
+                            ConfigOptions.DEFAULT_REPLICATION_FACTOR.key()));
+        }
+
+        if (conf.get(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS) < 1) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be greater than or equal 1.",
+                            ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS.key()));
+        }
+
+        if (conf.get(ConfigOptions.SERVER_IO_POOL_SIZE) < 1) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be greater than or equal 1.",
+                            ConfigOptions.SERVER_IO_POOL_SIZE.key()));
+        }
+
+        // validate remote.data.dirs
+        List<String> remoteDataDirs = conf.get(ConfigOptions.REMOTE_DATA_DIRS);
+        ConfigOptions.RemoteDataDirStrategy remoteDataDirStrategy =
+                conf.get(ConfigOptions.REMOTE_DATA_DIRS_STRATEGY);
+        if (remoteDataDirStrategy == ConfigOptions.RemoteDataDirStrategy.WEIGHTED_ROUND_ROBIN) {
+            List<Integer> weights = conf.get(ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS);
+            if (!remoteDataDirs.isEmpty() && !weights.isEmpty()) {
+                if (remoteDataDirs.size() != weights.size()) {
+                    throw new IllegalConfigurationException(
+                            String.format(
+                                    "The size of '%s' (%d) must match the size of '%s' (%d) when using WEIGHTED_ROUND_ROBIN strategy.",
+                                    ConfigOptions.REMOTE_DATA_DIRS.key(),
+                                    remoteDataDirs.size(),
+                                    ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS.key(),
+                                    weights.size()));
+                }
+                // validate all weights are positive
+                for (int i = 0; i < weights.size(); i++) {
+                    if (weights.get(i) < 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));
+                    }
+                }
+            }
+        }
+    }
+
+    public static void validateTabletConfigs(Configuration conf) {
+        validServerConfigs(conf);
+
+        Optional<Integer> serverId = conf.getOptional(ConfigOptions.TABLET_SERVER_ID);
+        if (!serverId.isPresent()) {
+            throw new IllegalConfigurationException(
+                    String.format("Configuration %s must be set.", ConfigOptions.TABLET_SERVER_ID));
+        }
+
+        if (serverId.get() < 0) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be greater than or equal 0.",
+                            ConfigOptions.TABLET_SERVER_ID.key()));
+        }
+
+        if (conf.get(ConfigOptions.BACKGROUND_THREADS) < 1) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be greater than or equal 1.",
+                            ConfigOptions.BACKGROUND_THREADS.key()));
+        }
+
+        if (conf.get(ConfigOptions.LOG_SEGMENT_FILE_SIZE).getBytes() > Integer.MAX_VALUE) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be less than or equal %d bytes.",
+                            ConfigOptions.LOG_SEGMENT_FILE_SIZE.key(), Integer.MAX_VALUE));
+        }
+    }
+
+    private static void validServerConfigs(Configuration conf) {
+        if (conf.get(ConfigOptions.REMOTE_DATA_DIR) == null) {
+            throw new IllegalConfigurationException(
+                    String.format("Configuration %s must be set.", ConfigOptions.REMOTE_DATA_DIR));
+        }
+    }
 }

fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.remote;
 
 import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 
@@ -50,14 +51,17 @@ public class RemoteLogSegment {
 
     private final int segmentSizeInBytes;
 
+    private final FsPath remoteLogDir;
+
     private RemoteLogSegment(
             PhysicalTablePath physicalTablePath,
             TableBucket tableBucket,
             UUID remoteLogSegmentId,
             long remoteLogStartOffset,
             long remoteLogEndOffset,
             long maxTimestamp,
-            int segmentSizeInBytes) {
+            int segmentSizeInBytes,
+            FsPath remoteLogDir) {
         this.physicalTablePath = checkNotNull(physicalTablePath);
         this.tableBucket = checkNotNull(tableBucket);
         this.remoteLogSegmentId = checkNotNull(remoteLogSegmentId);
@@ -79,6 +83,7 @@ private RemoteLogSegment(
         this.remoteLogEndOffset = remoteLogEndOffset;
         this.maxTimestamp = maxTimestamp;
         this.segmentSizeInBytes = segmentSizeInBytes;
+        this.remoteLogDir = remoteLogDir;
     }
 
     public PhysicalTablePath physicalTablePath() {
@@ -115,6 +120,10 @@ public int segmentSizeInBytes() {
         return segmentSizeInBytes;
     }
 
+    public FsPath remoteLogDir() {
+        return remoteLogDir;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -174,6 +183,7 @@ public static class Builder {
         private long remoteLogEndOffset;
         private long maxTimestamp;
         private int segmentSizeInBytes;
+        private FsPath remoteLogDir;
 
         public static Builder builder() {
             return new Builder();
@@ -214,6 +224,11 @@ public Builder tableBucket(TableBucket tableBucket) {
             return this;
         }
 
+        public Builder remoteLogDir(FsPath remoteLogDir) {
+            this.remoteLogDir = remoteLogDir;
+            return this;
+        }
+
         public RemoteLogSegment build() {
             return new RemoteLogSegment(
                     physicalTablePath,
@@ -222,7 +237,8 @@ public RemoteLogSegment build() {
                     remoteLogStartOffset,
                     remoteLogEndOffset,
                     maxTimestamp,
-                    segmentSizeInBytes);
+                    segmentSizeInBytes,
+                    remoteLogDir);
         }
     }
 }

fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java

@@ -401,6 +401,14 @@ public static UUID uuidFromRemoteIndexCacheFileName(String fileName) {
                 fileName.substring(fileName.indexOf('_') + 1, fileName.indexOf('.')));
     }
 
+    // ----------------------------------------------------------------------------------------
+    // Remote Data Paths
+    // ----------------------------------------------------------------------------------------
+
+    public static FsPath remoteDataDir(Configuration conf) {
+        return new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR));
+    }
+
     // ----------------------------------------------------------------------------------------
     // Remote Log Paths
     // ----------------------------------------------------------------------------------------
@@ -418,6 +426,10 @@ public static FsPath remoteLogDir(Configuration conf) {
         return new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR) + "/" + REMOTE_LOG_DIR_NAME);
     }
 
+    public static FsPath remoteLogDir(FsPath remoteDataDir) {
+        return new FsPath(remoteDataDir, REMOTE_LOG_DIR_NAME);
+    }
+
     /**
      * Returns the remote directory path for storing log files for a log tablet.
      *
@@ -584,6 +596,10 @@ public static FsPath remoteKvDir(Configuration conf) {
         return new FsPath(conf.get(ConfigOptions.REMOTE_DATA_DIR) + "/" + REMOTE_KV_DIR_NAME);
     }
 
+    public static FsPath remoteKvDir(FsPath remoteDataDir) {
+        return new FsPath(remoteDataDir, REMOTE_KV_DIR_NAME);
+    }
+
     /**
      * Returns the remote directory path for storing kv snapshot files for a kv tablet.
      *

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

@@ -17,14 +17,20 @@
 
 package org.apache.fluss.config;
 
+import org.apache.fluss.exception.IllegalConfigurationException;
+
 import org.junit.jupiter.api.Test;
 
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.Map;
 
 import static org.apache.fluss.config.FlussConfigUtils.CLIENT_OPTIONS;
 import static org.apache.fluss.config.FlussConfigUtils.TABLE_OPTIONS;
 import static org.apache.fluss.config.FlussConfigUtils.extractConfigOptions;
+import static org.apache.fluss.config.FlussConfigUtils.validateCoordinatorConfigs;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link FlussConfigUtils}. */
 class FlussConfigUtilsTest {
@@ -49,4 +55,72 @@ void testExtractOptions() {
                 });
         assertThat(clientOptions.size()).isEqualTo(CLIENT_OPTIONS.size());
     }
+
+    @Test
+    void testValidateCoordinatorConfigs() {
+        // Test valid configuration
+        Configuration validConf = new Configuration();
+        validConf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+        validateCoordinatorConfigs(validConf);
+
+        // Test invalid DEFAULT_REPLICATION_FACTOR
+        Configuration invalidReplicationConf = new Configuration();
+        invalidReplicationConf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+        invalidReplicationConf.set(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 0);
+        assertThatThrownBy(() -> validateCoordinatorConfigs(invalidReplicationConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining(ConfigOptions.DEFAULT_REPLICATION_FACTOR.key())
+                .hasMessageContaining("must be greater than or equal 1");
+
+        // Test invalid KV_MAX_RETAINED_SNAPSHOTS
+        Configuration invalidSnapshotConf = new Configuration();
+        invalidSnapshotConf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+        invalidSnapshotConf.set(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS, 0);
+        assertThatThrownBy(() -> validateCoordinatorConfigs(invalidSnapshotConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS.key())
+                .hasMessageContaining("must be greater than or equal 1");
+
+        // Test invalid SERVER_IO_POOL_SIZE
+        Configuration invalidIoPoolConf = new Configuration();
+        invalidIoPoolConf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+        invalidIoPoolConf.set(ConfigOptions.SERVER_IO_POOL_SIZE, 0);
+        assertThatThrownBy(() -> validateCoordinatorConfigs(invalidIoPoolConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining(ConfigOptions.SERVER_IO_POOL_SIZE.key())
+                .hasMessageContaining("must be greater than or equal 1");
+
+        // Test REMOTE_DATA_DIR not set
+        Configuration noRemoteDirConf = new Configuration();
+        assertThatThrownBy(() -> validateCoordinatorConfigs(noRemoteDirConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining(ConfigOptions.REMOTE_DATA_DIR.key())
+                .hasMessageContaining("must be set");
+
+        // Test WEIGHTED_ROUND_ROBIN with mismatched sizes
+        Configuration mismatchedWeightsConf = new Configuration();
+        mismatchedWeightsConf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+        mismatchedWeightsConf.set(
+                ConfigOptions.REMOTE_DATA_DIRS_STRATEGY,
+                ConfigOptions.RemoteDataDirStrategy.WEIGHTED_ROUND_ROBIN);
+        mismatchedWeightsConf.set(
+                ConfigOptions.REMOTE_DATA_DIRS, Arrays.asList("s3://bucket1", "s3://bucket2"));
+        mismatchedWeightsConf.set(
+                ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS, Collections.singletonList(1));
+        assertThatThrownBy(() -> validateCoordinatorConfigs(mismatchedWeightsConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining(ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS.key())
+                .hasMessageContaining(ConfigOptions.REMOTE_DATA_DIRS.key());
+
+        // Test WEIGHTED_ROUND_ROBIN with matched sizes
+        Configuration matchedWeightsConf = new Configuration();
+        matchedWeightsConf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+        matchedWeightsConf.set(
+                ConfigOptions.REMOTE_DATA_DIRS_STRATEGY,
+                ConfigOptions.RemoteDataDirStrategy.WEIGHTED_ROUND_ROBIN);
+        matchedWeightsConf.set(
+                ConfigOptions.REMOTE_DATA_DIRS, Arrays.asList("s3://bucket1", "s3://bucket2"));
+        matchedWeightsConf.set(ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS, Arrays.asList(1, 2));
+        validateCoordinatorConfigs(matchedWeightsConf);
+    }
 }