You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/05/10 16:31:27 UTC
(fluss) branch main updated: [server] Implement JBOD Phase 1 Local Multi-Directory Support (#3030)
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new d4cd1a277 [server] Implement JBOD Phase 1 Local Multi-Directory Support (#3030)
d4cd1a277 is described below
commit d4cd1a2771ff254a81877296a5cabfa966d102e2
Author: hanliu0830 <ha...@gmail.com>
AuthorDate: Mon May 11 00:31:24 2026 +0800
[server] Implement JBOD Phase 1 Local Multi-Directory Support (#3030)
---
.../org/apache/fluss/config/ConfigOptions.java | 18 +-
.../org/apache/fluss/server/TabletManagerBase.java | 63 +-
.../java/org/apache/fluss/server/kv/KvManager.java | 22 +-
.../org/apache/fluss/server/log/LogManager.java | 362 ++++++----
.../org/apache/fluss/server/log/LogTablet.java | 14 +
.../fluss/server/log/remote/RemoteLogManager.java | 62 +-
.../org/apache/fluss/server/replica/Replica.java | 12 +-
.../fluss/server/replica/ReplicaManager.java | 101 ++-
.../fluss/server/storage/LocalDiskManager.java | 747 +++++++++++++++++++++
.../apache/fluss/server/tablet/TabletServer.java | 24 +-
.../org/apache/fluss/server/kv/KvManagerTest.java | 34 +-
.../fluss/server/kv/KvTabletMergeModeTest.java | 1 +
.../server/kv/KvTabletSchemaEvolutionTest.java | 1 +
.../org/apache/fluss/server/kv/KvTabletTest.java | 1 +
.../fluss/server/kv/scan/ScannerManagerTest.java | 1 +
.../fluss/server/log/DroppedTableRecoveryTest.java | 92 ++-
.../org/apache/fluss/server/log/LogLoaderTest.java | 1 +
.../apache/fluss/server/log/LogManagerTest.java | 117 +++-
.../org/apache/fluss/server/log/LogTabletTest.java | 2 +
.../server/log/remote/RemoteLogIndexCacheTest.java | 2 +-
.../server/log/remote/RemoteLogManagerTest.java | 73 ++
.../fluss/server/replica/ReplicaManagerTest.java | 71 ++
.../fluss/server/replica/ReplicaTestBase.java | 44 +-
.../replica/fetcher/ReplicaFetcherThreadTest.java | 38 +-
.../fluss/server/storage/LocalDiskManagerTest.java | 400 +++++++++++
.../fluss/server/testutils/ServerTestTags.java | 26 +
26 files changed, 2047 insertions(+), 282 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index 02e016852..9f4a1145e 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -498,6 +498,20 @@ public class ConfigOptions {
"This configuration controls the directory where fluss will store its data. "
+ "The default value is /tmp/fluss-data");
+ public static final ConfigOption<List<String>> DATA_DIRS =
+ key("data.dirs")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "A comma-separated list of local directories used by TabletServer to store "
+ + "local log, kv, checkpoints, and other node-local files. "
+ + "If configured, this option takes precedence over `"
+ + DATA_DIR.key()
+ + "`. If not configured, `"
+ + DATA_DIR.key()
+ + "` is used as the only local data directory.");
+
public static final ConfigOption<Duration> WRITER_ID_EXPIRATION_TIME =
key("server.writer-id.expiration-time")
.durationType()
@@ -909,8 +923,8 @@ public class ConfigOptions {
.memoryType()
.defaultValue(MemorySize.parse("1gb"))
.withDescription(
- "The total size of the space allocated to store index files fetched "
- + "from remote storage in the local storage.");
+ "The size of the space allocated per local data directory to store "
+ + "index files fetched from remote storage in the local storage.");
public static final ConfigOption<Integer> REMOTE_LOG_MANAGER_THREAD_POOL_SIZE =
key("remote.log-manager.thread-pool-size")
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java b/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java
index 6525df1c4..5eb88df77 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java
@@ -43,12 +43,12 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
@@ -71,7 +71,7 @@ public abstract class TabletManagerBase {
KV
}
- protected final File dataDir;
+ protected final List<File> dataDirs;
protected final Configuration conf;
@@ -83,22 +83,31 @@ public abstract class TabletManagerBase {
private final String tabletDirPrefix;
public TabletManagerBase(
- TabletType tabletType, File dataDir, Configuration conf, int recoveryThreads) {
+ TabletType tabletType, List<File> dataDirs, Configuration conf, int recoveryThreads) {
this.tabletType = tabletType;
this.tabletDirPrefix = getTabletDirPrefix(tabletType);
- this.dataDir = dataDir;
+ this.dataDirs = new ArrayList<>(dataDirs);
this.conf = conf;
this.recoveryThreads = recoveryThreads;
}
/**
- * Return the directories of the tablets to be loaded.
+ * Return the directories of the tablets to be loaded, grouped by configured data directory.
*
* <p>See more about the local directory contracts: {@link FlussPaths#logTabletDir(File,
* PhysicalTablePath, TableBucket)} and {@link FlussPaths#kvTabletDir(File, PhysicalTablePath,
* TableBucket)}.
*/
- protected List<File> listTabletsToLoad() {
+ protected Map<File, List<File>> listTabletsToLoad() {
+ Map<File, List<File>> tabletsToLoadByDataDir = new LinkedHashMap<>();
+ for (File dataDir : dataDirs) {
+ tabletsToLoadByDataDir.put(dataDir, listTabletsToLoad(dataDir));
+ }
+ return tabletsToLoadByDataDir;
+ }
+
+ /** Returns the tablet directories to be loaded from a single configured data directory. */
+ protected List<File> listTabletsToLoad(File dataDir) {
List<File> tabletsToLoad = new ArrayList<>();
// Get all database directory.
File[] dbDirs = FileUtils.listDirectories(dataDir);
@@ -140,41 +149,21 @@ public abstract class TabletManagerBase {
return Executors.newFixedThreadPool(recoveryThreads, new ExecutorThreadFactory(poolName));
}
- /** Running a series of jobs in a thread pool, and return the count of the successful job. */
- protected int runInThreadPool(Runnable[] runnableJobs, String poolName) throws Throwable {
- List<Future<?>> jobsForTabletDir = new ArrayList<>();
- ExecutorService pool = createThreadPool(poolName);
- for (Runnable runnable : runnableJobs) {
- jobsForTabletDir.add(pool.submit(runnable));
- }
- int successCount = 0;
- try {
- for (Future<?> future : jobsForTabletDir) {
- try {
- future.get();
- successCount++;
- } catch (InterruptedException | ExecutionException e) {
- throw e.getCause();
- }
- }
- } finally {
- pool.shutdown();
- }
- return successCount;
- }
-
/**
- * Get the tablet directory with given directory name for the given table path and table bucket.
+ * Get the tablet directory with given directory name for the given data directory, table path
+ * and table bucket.
*
* <p>When the parent directory of the tablet directory is missing, it will create the
* directory.
*
+ * @param dataDir the local data directory chosen for this tablet
* @param tablePath the table path of the bucket
* @param tableBucket the table bucket
* @return the tablet directory
*/
- protected File getOrCreateTabletDir(PhysicalTablePath tablePath, TableBucket tableBucket) {
- File tabletDir = getTabletDir(tablePath, tableBucket);
+ protected File getOrCreateTabletDir(
+ File dataDir, PhysicalTablePath tablePath, TableBucket tableBucket) {
+ File tabletDir = getTabletDir(dataDir, tablePath, tableBucket);
if (tabletDir.exists()) {
return tabletDir;
}
@@ -182,11 +171,13 @@ public abstract class TabletManagerBase {
return tabletDir;
}
- public Path getTabletParentDir(PhysicalTablePath tablePath, TableBucket tableBucket) {
- return getTabletDir(tablePath, tableBucket).toPath().getParent();
+ public Path getTabletParentDir(
+ File dataDir, PhysicalTablePath tablePath, TableBucket tableBucket) {
+ return getTabletDir(dataDir, tablePath, tableBucket).toPath().getParent();
}
- protected File getTabletDir(PhysicalTablePath tablePath, TableBucket tableBucket) {
+ protected File getTabletDir(
+ File dataDir, PhysicalTablePath tablePath, TableBucket tableBucket) {
switch (tabletType) {
case LOG:
return FlussPaths.logTabletDir(dataDir, tablePath, tableBucket);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
index 14617ae1c..9f2f1f426 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
@@ -42,6 +42,7 @@ import org.apache.fluss.server.kv.rowmerger.RowMerger;
import org.apache.fluss.server.log.LogManager;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BufferAllocator;
import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BufferAllocatorUtil;
@@ -110,6 +111,7 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
}
private final LogManager logManager;
+ private final LocalDiskManager localDiskManager;
private final TabletServerMetricGroup serverMetricGroup;
@@ -141,14 +143,15 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
private volatile boolean isShutdown = false;
private KvManager(
- File dataDir,
+ LocalDiskManager localDiskManager,
Configuration conf,
ZooKeeperClient zkClient,
int recoveryThreadsPerDataDir,
LogManager logManager,
TabletServerMetricGroup tabletServerMetricGroup)
throws IOException {
- super(TabletType.KV, dataDir, conf, recoveryThreadsPerDataDir);
+ super(TabletType.KV, localDiskManager.dataDirs(), conf, recoveryThreadsPerDataDir);
+ this.localDiskManager = localDiskManager;
this.logManager = logManager;
this.arrowBufferAllocator = BufferAllocatorUtil.createBufferAllocator(null);
this.memorySegmentPool = LazyMemorySegmentPool.createServerBufferPool(conf);
@@ -183,12 +186,11 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
Configuration conf,
ZooKeeperClient zkClient,
LogManager logManager,
- TabletServerMetricGroup tabletServerMetricGroup)
+ TabletServerMetricGroup tabletServerMetricGroup,
+ LocalDiskManager localDiskManager)
throws IOException {
- String dataDirString = conf.getString(ConfigOptions.DATA_DIR);
- File dataDir = new File(dataDirString).getAbsoluteFile();
return new KvManager(
- dataDir,
+ localDiskManager,
conf,
zkClient,
conf.getInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS),
@@ -248,7 +250,8 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
return currentKvs.get(tableBucket);
}
- File tabletDir = getOrCreateTabletDir(tablePath, tableBucket);
+ File tabletDir =
+ getOrCreateTabletDir(logTablet.getDataDir(), tablePath, tableBucket);
RowMerger merger = RowMerger.create(tableConfig, kvFormat, schemaGetter);
AutoIncrementManager autoIncrementManager =
new AutoIncrementManager(
@@ -294,8 +297,9 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
* @param tableBucket the table bucket
* @return the tablet directory
*/
- public File createTabletDir(PhysicalTablePath tablePath, TableBucket tableBucket) {
- File tabletDir = getTabletDir(tablePath, tableBucket);
+ public File createTabletDir(
+ File dataDir, PhysicalTablePath tablePath, TableBucket tableBucket) {
+ File tabletDir = getTabletDir(dataDir, tablePath, tableBucket);
// delete the tablet dir if exists
FileUtils.deleteDirectoryQuietly(tabletDir);
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java
index f7f35f461..ae0432820 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java
@@ -20,6 +20,7 @@ package org.apache.fluss.server.log;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.LogStorageException;
import org.apache.fluss.exception.SchemaNotExistException;
@@ -31,7 +32,9 @@ import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.TabletManagerBase;
import org.apache.fluss.server.log.checkpoint.OffsetCheckpointFile;
import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.utils.ExceptionUtils;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.FlussPaths;
import org.apache.fluss.utils.clock.Clock;
@@ -46,9 +49,9 @@ import javax.annotation.concurrent.ThreadSafe;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
-import java.nio.file.Path;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -57,6 +60,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
import static org.apache.fluss.utils.concurrent.LockUtils.inLock;
@@ -90,15 +94,16 @@ public final class LogManager extends TabletManagerBase {
private final Scheduler scheduler;
private final Clock clock;
private final TabletServerMetricGroup serverMetricGroup;
+ private final LocalDiskManager localDiskManager;
private final ReentrantLock logCreationOrDeletionLock = new ReentrantLock();
private final Map<TableBucket, LogTablet> currentLogs = new ConcurrentHashMap<>();
- private volatile OffsetCheckpointFile recoveryPointCheckpoint;
+ private volatile Map<File, OffsetCheckpointFile> recoveryPointCheckpoints;
private boolean loadLogsCompletedFlag = false;
private LogManager(
- File dataDir,
+ LocalDiskManager localDiskManager,
Configuration conf,
ZooKeeperClient zkClient,
int recoveryThreadsPerDataDir,
@@ -106,12 +111,12 @@ public final class LogManager extends TabletManagerBase {
Clock clock,
TabletServerMetricGroup serverMetricGroup)
throws Exception {
- super(TabletType.LOG, dataDir, conf, recoveryThreadsPerDataDir);
+ super(TabletType.LOG, localDiskManager.dataDirs(), conf, recoveryThreadsPerDataDir);
this.zkClient = zkClient;
this.scheduler = scheduler;
this.clock = clock;
this.serverMetricGroup = serverMetricGroup;
- createAndValidateDataDir(dataDir);
+ this.localDiskManager = localDiskManager;
initializeCheckpointMaps();
}
@@ -121,12 +126,11 @@ public final class LogManager extends TabletManagerBase {
ZooKeeperClient zkClient,
Scheduler scheduler,
Clock clock,
- TabletServerMetricGroup serverMetricGroup)
+ TabletServerMetricGroup serverMetricGroup,
+ LocalDiskManager localDiskManager)
throws Exception {
- String dataDirString = conf.getString(ConfigOptions.DATA_DIR);
- File dataDir = new File(dataDirString).getAbsoluteFile();
return new LogManager(
- dataDir,
+ localDiskManager,
conf,
zkClient,
conf.getInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS),
@@ -136,76 +140,116 @@ public final class LogManager extends TabletManagerBase {
}
public void startup() {
- loadLogs();
+ loadAllLogs();
// TODO add more scheduler, like log-flusher etc.
}
- public File getDataDir() {
- return dataDir;
- }
-
private void initializeCheckpointMaps() throws IOException {
- recoveryPointCheckpoint =
- new OffsetCheckpointFile(new File(dataDir, RECOVERY_POINT_CHECKPOINT_FILE));
+ recoveryPointCheckpoints = new HashMap<>();
+ for (File dataDir : dataDirs) {
+ recoveryPointCheckpoints.put(
+ dataDir,
+ new OffsetCheckpointFile(new File(dataDir, RECOVERY_POINT_CHECKPOINT_FILE)));
+ }
}
/** Recover and load all logs in the given data directories. */
- private void loadLogs() {
- LOG.info("Loading logs from dir {}", dataDir);
-
- String dataDirAbsolutePath = dataDir.getAbsolutePath();
+ private void loadAllLogs() {
try {
- boolean isCleanShutdown = false;
- File cleanShutdownFile = new File(dataDir, CLEAN_SHUTDOWN_FILE);
- if (cleanShutdownFile.exists()) {
- // Cache the clean shutdown status marker and use that for rest of log loading
- // workflow. Delete the CleanShutdownFile so that if tabletServer crashes while
- // loading the log, it is considered hard shutdown during the next boot up.
- Files.deleteIfExists(cleanShutdownFile.toPath());
- isCleanShutdown = true;
+ Map<File, List<File>> tabletsToLoadByDataDir = listTabletsToLoad();
+ LOG.info("Loading logs from {} dirs", tabletsToLoadByDataDir.size());
+ List<LogRecoveryTask> recoveryTasks = new ArrayList<>();
+ for (Map.Entry<File, List<File>> entry : tabletsToLoadByDataDir.entrySet()) {
+ recoveryTasks.add(loadLogsInDir(entry.getKey(), entry.getValue()));
}
-
- Map<TableBucket, Long> recoveryPoints = new HashMap<>();
try {
- recoveryPoints = recoveryPointCheckpoint.read();
- } catch (Exception e) {
- LOG.warn(
- "Error occurred while reading recovery-point-offset-checkpoint file of directory {}, "
- + "resetting the recovery checkpoint to 0",
- dataDirAbsolutePath,
- e);
+ for (LogRecoveryTask recoveryTask : recoveryTasks) {
+ waitForLoadLogsInDir(recoveryTask);
+ }
+ loadLogsCompletedFlag = true;
+ LOG.info("Log loader complete.");
+ } finally {
+ for (LogRecoveryTask recoveryTask : recoveryTasks) {
+ recoveryTask.pool.shutdown();
+ }
}
+ } catch (Throwable e) {
+ throw new FlussRuntimeException("Failed to recover logs", e);
+ }
+ }
- List<File> tabletsToLoad = listTabletsToLoad();
- if (tabletsToLoad.isEmpty()) {
- LOG.info("No logs found to be loaded in {}", dataDirAbsolutePath);
- } else if (isCleanShutdown) {
- LOG.info("Skipping some recovery log process since clean shutdown file was found");
- } else {
- LOG.info("Recovering all local logs since no clean shutdown file was found");
- }
+ private LogRecoveryTask loadLogsInDir(File dataDir, List<File> tabletsToLoad) throws Exception {
+ LOG.info("Loading logs from dir {}", dataDir);
+
+ String dataDirAbsolutePath = dataDir.getAbsolutePath();
+ boolean isCleanShutdown = false;
+ File cleanShutdownFile = new File(dataDir, CLEAN_SHUTDOWN_FILE);
+ if (cleanShutdownFile.exists()) {
+ // Cache the clean shutdown status marker and use that for rest of log loading
+ // workflow. Delete the CleanShutdownFile so that if tabletServer crashes while
+ // loading the log, it is considered hard shutdown during the next boot up.
+ Files.deleteIfExists(cleanShutdownFile.toPath());
+ isCleanShutdown = true;
+ }
- final Map<TableBucket, Long> finalRecoveryPoints = recoveryPoints;
- final boolean cleanShutdown = isCleanShutdown;
- // set runnable job.
- Runnable[] jobsForDir =
- createLogLoadingJobs(
- tabletsToLoad, cleanShutdown, finalRecoveryPoints, conf, clock);
+ Map<TableBucket, Long> recoveryPoints = new HashMap<>();
+ try {
+ recoveryPoints = recoveryPointCheckpoints.get(dataDir).read();
+ } catch (Exception e) {
+ LOG.warn(
+ "Error occurred while reading recovery-point-offset-checkpoint file of directory {}, "
+ + "resetting the recovery checkpoint to 0",
+ dataDirAbsolutePath,
+ e);
+ }
- long startTime = System.currentTimeMillis();
+ if (tabletsToLoad.isEmpty()) {
+ LOG.info("No logs found to be loaded in {}", dataDirAbsolutePath);
+ } else if (isCleanShutdown) {
+ LOG.info("Skipping some recovery log process since clean shutdown file was found");
+ } else {
+ LOG.info("Recovering all local logs since no clean shutdown file was found");
+ }
- int successLoadCount =
- runInThreadPool(jobsForDir, "log-recovery-" + dataDirAbsolutePath);
+ final Map<TableBucket, Long> finalRecoveryPoints = recoveryPoints;
+ final boolean cleanShutdown = isCleanShutdown;
+ // set runnable job.
+ Runnable[] jobsForDir =
+ createLogLoadingJobs(
+ dataDir, tabletsToLoad, cleanShutdown, finalRecoveryPoints, conf, clock);
+
+ long startTimeMillis = System.currentTimeMillis();
+ List<Future<?>> jobsForDataDir = new ArrayList<>();
+ ExecutorService pool = createThreadPool("log-recovery-" + dataDirAbsolutePath);
+ for (Runnable job : jobsForDir) {
+ jobsForDataDir.add(pool.submit(job));
+ }
+ return new LogRecoveryTask(dataDir, pool, jobsForDataDir, startTimeMillis);
+ }
- loadLogsCompletedFlag = true;
- LOG.info(
- "Log loader complete. Total success loaded log count is {}, Take {} ms",
- successLoadCount,
- System.currentTimeMillis() - startTime);
- } catch (Throwable e) {
- throw new FlussRuntimeException("Failed to recovery log", e);
+ private void waitForLoadLogsInDir(LogRecoveryTask recoveryTask) throws Throwable {
+ int successCount = 0;
+ for (Future<?> future : recoveryTask.jobs) {
+ try {
+ future.get();
+ successCount++;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new FlussRuntimeException(
+ "Interrupted while waiting for log recovery tasks to finish.", e);
+ } catch (ExecutionException e) {
+ throw new FlussException(
+ "Failed while waiting for log recovery tasks to finish.",
+ ExceptionUtils.stripExecutionException(e));
+ }
}
+
+ LOG.info(
+ "Log loader complete for {}. Total success loaded log count is {}, Take {} ms",
+ recoveryTask.dataDir.getAbsolutePath(),
+ successCount,
+ System.currentTimeMillis() - recoveryTask.startTimeMillis);
}
/**
@@ -213,6 +257,7 @@ public final class LogManager extends TabletManagerBase {
* return a copy of the existing log. Otherwise, create a log for the given table and the given
* bucket.
*
+ * @param dataDir the local data directory chosen for the bucket
* @param tablePath the table path of the bucket belongs to
* @param tableBucket the table bucket
* @param logFormat the log format
@@ -220,6 +265,7 @@ public final class LogManager extends TabletManagerBase {
* @param isChangelog whether the log is a changelog of primary key table
*/
public LogTablet getOrCreateLog(
+ File dataDir,
PhysicalTablePath tablePath,
TableBucket tableBucket,
LogFormat logFormat,
@@ -233,10 +279,11 @@ public final class LogManager extends TabletManagerBase {
return currentLogs.get(tableBucket);
}
- File tabletDir = getOrCreateTabletDir(tablePath, tableBucket);
+ File tabletDir = getOrCreateTabletDir(dataDir, tablePath, tableBucket);
LogTablet logTablet =
LogTablet.create(
+ dataDir,
tablePath,
tabletDir,
conf,
@@ -311,7 +358,7 @@ public final class LogManager extends TabletManagerBase {
LogTablet logTablet = currentLogs.get(tableBucket);
// If the log tablet does not exist, skip it.
if (logTablet != null && logTablet.truncateTo(offset)) {
- checkpointRecoveryOffsets();
+ checkpointRecoveryOffsets(logTablet.getDataDir());
}
}
@@ -320,11 +367,12 @@ public final class LogManager extends TabletManagerBase {
// If the log tablet does not exist, skip it.
if (logTablet != null) {
logTablet.truncateFullyAndStartAt(newOffset);
- checkpointRecoveryOffsets();
+ checkpointRecoveryOffsets(logTablet.getDataDir());
}
}
private LogTablet loadLog(
+ File dataDir,
File tabletDir,
boolean isCleanShutdown,
Map<TableBucket, Long> recoveryPoints,
@@ -340,6 +388,7 @@ public final class LogManager extends TabletManagerBase {
TableInfo tableInfo = getTableInfo(zkClient, tablePath);
LogTablet logTablet =
LogTablet.create(
+ dataDir,
physicalTablePath,
tabletDir,
conf,
@@ -367,50 +416,48 @@ public final class LogManager extends TabletManagerBase {
currentLogs.get(tableBucket).getLogDir().getAbsolutePath()));
}
currentLogs.put(tableBucket, logTablet);
+ localDiskManager.recordReplicaLoad(dataDir, tableInfo.hasPrimaryKey());
return logTablet;
}
- private void createAndValidateDataDir(File dataDir) {
- try {
- inLock(
- logCreationOrDeletionLock,
- () -> {
- if (!dataDir.exists()) {
- LOG.info(
- "Data directory {} not found, creating it.",
- dataDir.getAbsolutePath());
- boolean created = dataDir.mkdirs();
- if (!created) {
- throw new IOException(
- "Failed to create data directory "
- + dataDir.getAbsolutePath());
- }
- Path parentPath =
- dataDir.toPath().toAbsolutePath().normalize().getParent();
- FileUtils.flushDir(parentPath);
- }
- if (!dataDir.isDirectory() || !dataDir.canRead()) {
- throw new IOException(
- dataDir.getAbsolutePath()
- + " is not a readable data directory.");
- }
- });
- } catch (IOException e) {
- throw new FlussRuntimeException(
- "Failed to create or validate data directory " + dataDir.getAbsolutePath(), e);
- }
- }
-
/** Close all the logs. */
public void shutdown() {
LOG.info("Shutting down LogManager.");
+ Map<File, List<LogTablet>> logsByDataDir = new LinkedHashMap<>();
+ for (File dataDir : dataDirs) {
+ logsByDataDir.put(dataDir, new ArrayList<>());
+ }
+ for (LogTablet logTablet : currentLogs.values()) {
+ File dataDir = logTablet.getDataDir();
+ logsByDataDir.computeIfAbsent(dataDir, ignored -> new ArrayList<>()).add(logTablet);
+ }
+
+ List<LogShutdownTask> shutdownTasks = new ArrayList<>();
+ for (Map.Entry<File, List<LogTablet>> entry : logsByDataDir.entrySet()) {
+ shutdownTasks.add(shutdownLogsInDir(entry.getKey(), entry.getValue()));
+ }
+
+ try {
+ for (LogShutdownTask shutdownTask : shutdownTasks) {
+ waitForShutdownLogsInDir(shutdownTask);
+ }
+ } finally {
+ for (LogShutdownTask shutdownTask : shutdownTasks) {
+ shutdownTask.pool.shutdown();
+ }
+ }
+
+ LOG.info("Shut down LogManager complete.");
+ }
+
+ private LogShutdownTask shutdownLogsInDir(File dataDir, List<LogTablet> logs) {
String dataDirAbsolutePath = dataDir.getAbsolutePath();
- ExecutorService pool = createThreadPool("log-tablet-closing-" + dataDirAbsolutePath);
+ LOG.info("Shutting down {} logs in dir {}", logs.size(), dataDirAbsolutePath);
- List<LogTablet> logs = new ArrayList<>(currentLogs.values());
List<Future<?>> jobsForTabletDir = new ArrayList<>();
+ ExecutorService pool = createThreadPool("log-tablet-closing-" + dataDirAbsolutePath);
for (LogTablet logTablet : logs) {
Runnable runnable =
() -> {
@@ -423,42 +470,40 @@ public final class LogManager extends TabletManagerBase {
};
jobsForTabletDir.add(pool.submit(runnable));
}
+ return new LogShutdownTask(dataDir, logs, pool, jobsForTabletDir);
+ }
- try {
- for (Future<?> future : jobsForTabletDir) {
- try {
- future.get();
- } catch (InterruptedException e) {
- LOG.warn("Interrupted while shutting down LogManager.");
- } catch (ExecutionException e) {
- LOG.warn(
- "There was an error in one of the threads during LogManager shutdown",
- e);
- }
+ private void waitForShutdownLogsInDir(LogShutdownTask shutdownTask) {
+ for (Future<?> future : shutdownTask.jobs) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while shutting down LogManager.");
+ } catch (ExecutionException e) {
+ LOG.warn("There was an error in one of the threads during LogManager shutdown", e);
}
+ }
- // update the last flush point.
- checkpointRecoveryOffsets();
+ checkpointRecoveryOffsets(shutdownTask.dataDir, shutdownTask.logs);
- // mark that the shutdown was clean by creating marker file for log dirs that all logs
- // have been recovered at startup time.
- if (loadLogsCompletedFlag) {
- LOG.debug("Writing clean shutdown marker.");
- try {
- Files.createFile(new File(dataDir, CLEAN_SHUTDOWN_FILE).toPath());
- } catch (IOException e) {
- LOG.warn("Failed to write clean shutdown marker.", e);
- }
+ // mark that the shutdown was clean by creating marker file for log dirs that all logs have
+ // been recovered at startup time.
+ if (loadLogsCompletedFlag) {
+ try {
+ LOG.debug("Writing clean shutdown marker for directory {}.", shutdownTask.dataDir);
+ Files.createFile(new File(shutdownTask.dataDir, CLEAN_SHUTDOWN_FILE).toPath());
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to write clean shutdown marker for directory {}.",
+ shutdownTask.dataDir,
+ e);
}
- } finally {
- pool.shutdown();
}
-
- LOG.info("Shut down LogManager complete.");
}
/** Create runnable jobs for loading logs from tablet directories. */
private Runnable[] createLogLoadingJobs(
+ File dataDir,
List<File> tabletsToLoad,
boolean cleanShutdown,
Map<TableBucket, Long> recoveryPoints,
@@ -467,13 +512,16 @@ public final class LogManager extends TabletManagerBase {
Runnable[] jobs = new Runnable[tabletsToLoad.size()];
for (int i = 0; i < tabletsToLoad.size(); i++) {
final File tabletDir = tabletsToLoad.get(i);
- jobs[i] = createLogLoadingJob(tabletDir, cleanShutdown, recoveryPoints, conf, clock);
+ jobs[i] =
+ createLogLoadingJob(
+ dataDir, tabletDir, cleanShutdown, recoveryPoints, conf, clock);
}
return jobs;
}
/** Create a runnable job for loading log from a single tablet directory. */
private Runnable createLogLoadingJob(
+ File dataDir,
File tabletDir,
boolean cleanShutdown,
Map<TableBucket, Long> recoveryPoints,
@@ -484,7 +532,7 @@ public final class LogManager extends TabletManagerBase {
public void run() {
LOG.debug("Loading log {}", tabletDir);
try {
- loadLog(tabletDir, cleanShutdown, recoveryPoints, conf, clock);
+ loadLog(dataDir, tabletDir, cleanShutdown, recoveryPoints, conf, clock);
} catch (Exception e) {
LOG.error("Fail to loadLog from {}", tabletDir, e);
if (e instanceof SchemaNotExistException) {
@@ -522,23 +570,55 @@ public final class LogManager extends TabletManagerBase {
}
@VisibleForTesting
- void checkpointRecoveryOffsets() {
- // Assuming TableBucket and LogTablet are actual types used in your application
- if (recoveryPointCheckpoint != null) {
- try {
- Map<TableBucket, Long> recoveryOffsets = new HashMap<>();
- for (Map.Entry<TableBucket, LogTablet> entry : currentLogs.entrySet()) {
- recoveryOffsets.put(entry.getKey(), entry.getValue().getRecoveryPoint());
- }
- recoveryPointCheckpoint.write(recoveryOffsets);
- } catch (Exception e) {
- throw new LogStorageException(
- "Disk error while writing recovery offsets checkpoint in directory "
- + dataDir
- + ": "
- + e.getMessage(),
- e);
+ void checkpointRecoveryOffsets(File dataDir) {
+ checkpointRecoveryOffsets(
+ dataDir,
+ currentLogs.values().stream()
+ .filter(log -> log.getDataDir().equals(dataDir))
+ .collect(Collectors.toList()));
+ }
+
+ private void checkpointRecoveryOffsets(File dataDir, List<LogTablet> logs) {
+ try {
+ Map<TableBucket, Long> recoveryOffsets = new HashMap<>();
+ for (LogTablet logTablet : logs) {
+ recoveryOffsets.put(logTablet.getTableBucket(), logTablet.getRecoveryPoint());
}
+ LOG.debug("Writing recovery offsets checkpoint for directory {}.", dataDir);
+ recoveryPointCheckpoints.get(dataDir).write(recoveryOffsets);
+ } catch (Exception e) {
+ throw new LogStorageException(
+ "Disk error while writing recovery offsets checkpoint", e);
+ }
+ }
+
+ private static final class LogRecoveryTask {
+ private final File dataDir;
+ private final ExecutorService pool;
+ private final List<Future<?>> jobs;
+ private final long startTimeMillis;
+
+ private LogRecoveryTask(
+ File dataDir, ExecutorService pool, List<Future<?>> jobs, long startTimeMillis) {
+ this.dataDir = dataDir;
+ this.pool = pool;
+ this.jobs = jobs;
+ this.startTimeMillis = startTimeMillis;
+ }
+ }
+
+ private static final class LogShutdownTask {
+ private final File dataDir;
+ private final List<LogTablet> logs;
+ private final ExecutorService pool;
+ private final List<Future<?>> jobs;
+
+ private LogShutdownTask(
+ File dataDir, List<LogTablet> logs, ExecutorService pool, List<Future<?>> jobs) {
+ this.dataDir = dataDir;
+ this.logs = logs;
+ this.pool = pool;
+ this.jobs = jobs;
}
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
index c77dd66b6..23c60c82e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
@@ -84,8 +84,12 @@ public final class LogTablet {
private static final Logger LOG = LoggerFactory.getLogger(LogTablet.class);
+ // Configured local storage root that owns this tablet, for example /data-0.
+ private final File dataDir;
+ // Logical table/partition identity of this tablet.
private final PhysicalTablePath physicalPath;
+ // Concrete log tablet directory under dataDir, for example /data-0/db/table/log-0.
@GuardedBy("lock")
private final LocalLog localLog;
@@ -130,6 +134,7 @@ public final class LogTablet {
private volatile long lakeMaxTimestamp = -1;
private LogTablet(
+ File dataDir,
PhysicalTablePath physicalPath,
LocalLog localLog,
Configuration conf,
@@ -139,6 +144,7 @@ public final class LogTablet {
int tieredLogLocalSegments,
boolean isChangelog,
Clock clock) {
+ this.dataDir = dataDir;
this.physicalPath = physicalPath;
this.localLog = localLog;
this.maxSegmentFileSize = (int) conf.get(ConfigOptions.LOG_SEGMENT_FILE_SIZE).getBytes();
@@ -221,6 +227,12 @@ public final class LogTablet {
}
}
+ /** Returns the configured local data directory that owns this tablet. */
+ public File getDataDir() {
+ return dataDir;
+ }
+
+ /** Returns the concrete log tablet directory under the owning local data directory. */
public File getLogDir() {
return localLog.getLogTabletDir();
}
@@ -291,6 +303,7 @@ public final class LogTablet {
}
public static LogTablet create(
+ File dataDir,
PhysicalTablePath tablePath,
File tabletDir,
Configuration conf,
@@ -339,6 +352,7 @@ public final class LogTablet {
logFormat);
return new LogTablet(
+ dataDir,
tablePath,
log,
conf,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java
index 282ae8d9e..143ae251a 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java
@@ -20,14 +20,17 @@ package org.apache.fluss.server.log.remote;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.NotLeaderOrFollowerException;
import org.apache.fluss.exception.RemoteStorageException;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
+import org.apache.fluss.server.log.LogManager;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.replica.Replica;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.RemoteLogManifestHandle;
import org.apache.fluss.utils.IOUtils;
@@ -71,12 +74,13 @@ public class RemoteLogManager implements Closeable {
private final long taskInterval;
private final int maxUploadSegmentsPerTask;
- private final RemoteLogIndexCache remoteLogIndexCache;
+ private final Map<File, RemoteLogIndexCache> remoteLogIndexCachesByDir;
private final RemoteLogStorage remoteLogStorage;
private final CoordinatorGateway coordinatorGateway;
private final ScheduledExecutorService rlManagerScheduledThreadPool;
private final Clock clock;
private final ZooKeeperClient zkClient;
+ private final LogManager logManager;
private final Map<TableBucket, TaskWithFuture> rlmTasks = new ConcurrentHashMap<>();
private final Map<TableBucket, RemoteLogTablet> remoteLogs = new ConcurrentHashMap<>();
@@ -85,6 +89,8 @@ public class RemoteLogManager implements Closeable {
Configuration conf,
ZooKeeperClient zkClient,
CoordinatorGateway coordinatorGateway,
+ LocalDiskManager localDiskManager,
+ LogManager logManager,
Clock clock,
ExecutorService ioExecutor)
throws IOException {
@@ -92,6 +98,8 @@ public class RemoteLogManager implements Closeable {
conf,
zkClient,
coordinatorGateway,
+ localDiskManager,
+ logManager,
new DefaultRemoteLogStorage(conf, ioExecutor),
Executors.newScheduledThreadPool(
conf.getInt(ConfigOptions.REMOTE_LOG_MANAGER_THREAD_POOL_SIZE),
@@ -104,6 +112,8 @@ public class RemoteLogManager implements Closeable {
Configuration conf,
ZooKeeperClient zkClient,
CoordinatorGateway coordinatorGateway,
+ LocalDiskManager localDiskManager,
+ LogManager logManager,
RemoteLogStorage remoteLogStorage,
ScheduledExecutorService scheduledExecutor,
Clock clock)
@@ -111,13 +121,13 @@ public class RemoteLogManager implements Closeable {
this.remoteLogStorage = remoteLogStorage;
this.zkClient = zkClient;
this.coordinatorGateway = coordinatorGateway;
-
- File dataDir = new File(conf.getString(ConfigOptions.DATA_DIR));
- this.remoteLogIndexCache =
- new RemoteLogIndexCache(
- (int) conf.get(ConfigOptions.REMOTE_LOG_INDEX_FILE_CACHE_SIZE).getBytes(),
- remoteLogStorage,
- dataDir);
+ this.logManager = logManager;
+ this.remoteLogIndexCachesByDir = new ConcurrentHashMap<>();
+ int cacheSize = (int) conf.get(ConfigOptions.REMOTE_LOG_INDEX_FILE_CACHE_SIZE).getBytes();
+ for (File dataDir : localDiskManager.dataDirs()) {
+ remoteLogIndexCachesByDir.put(
+ dataDir, new RemoteLogIndexCache(cacheSize, remoteLogStorage, dataDir));
+ }
this.taskInterval = conf.get(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION).toMillis();
this.maxUploadSegmentsPerTask =
conf.getInt(ConfigOptions.REMOTE_LOG_TASK_MAX_UPLOAD_SEGMENTS);
@@ -191,7 +201,8 @@ public class RemoteLogManager implements Closeable {
.map(RemoteLogSegment::remoteLogSegmentId)
.collect(Collectors.toList());
// remove cache.
- remoteLogIndexCache.removeAll(remoteLogSegmentIdList);
+ remoteLogIndexCache(replica.getLogTablet().getDataDir())
+ .removeAll(remoteLogSegmentIdList);
// unregister the remote log metrics, only leader needs to report
remoteLog.unregisterMetrics();
}
@@ -228,7 +239,8 @@ public class RemoteLogManager implements Closeable {
/** Get the position of the given offset in the remote log segment. */
public int lookupPositionForOffset(RemoteLogSegment remoteLogSegment, long offset) {
- return remoteLogIndexCache.lookupPosition(remoteLogSegment, offset);
+ return remoteLogIndexCacheForBucket(remoteLogSegment.tableBucket())
+ .lookupPosition(remoteLogSegment, offset);
}
/**
@@ -249,7 +261,8 @@ public class RemoteLogManager implements Closeable {
if (segment == null) {
return -1L;
} else {
- return remoteLogIndexCache.lookupOffsetForTimestamp(segment, timestamp);
+ return remoteLogIndexCacheForBucket(tableBucket)
+ .lookupOffsetForTimestamp(segment, timestamp);
}
}
@@ -332,7 +345,9 @@ public class RemoteLogManager implements Closeable {
public void close() throws IOException {
rlmTasks.values().forEach(TaskWithFuture::cancel);
IOUtils.closeQuietly(remoteLogStorage, "RemoteLogStorageManager");
- IOUtils.closeQuietly(remoteLogIndexCache, "RemoteIndexCache");
+ remoteLogIndexCachesByDir.forEach(
+ (dataDir, cache) ->
+ IOUtils.closeQuietly(cache, "RemoteIndexCache-" + dataDir.getName()));
shutdownAndAwaitTermination(
rlManagerScheduledThreadPool, "RLMScheduledThreadPool", 10, TimeUnit.SECONDS);
@@ -398,8 +413,8 @@ public class RemoteLogManager implements Closeable {
}
@VisibleForTesting
- public RemoteLogIndexCache getRemoteLogIndexCache() {
- return remoteLogIndexCache;
+ public RemoteLogIndexCache getRemoteLogIndexCache(File dataDir) {
+ return remoteLogIndexCache(dataDir);
}
@VisibleForTesting
@@ -407,4 +422,23 @@ public class RemoteLogManager implements Closeable {
TaskWithFuture getTaskWithFuture(TableBucket tableBucket) {
return rlmTasks.get(tableBucket);
}
+
+ private RemoteLogIndexCache remoteLogIndexCacheForBucket(TableBucket tableBucket) {
+ Optional<LogTablet> logTabletOpt = logManager.getLog(tableBucket);
+ if (!logTabletOpt.isPresent()) {
+ throw new NotLeaderOrFollowerException(
+ String.format(
+ "Can't resolve remote log index cache for bucket %s because no local log exists.",
+ tableBucket));
+ }
+ return remoteLogIndexCache(logTabletOpt.get().getDataDir());
+ }
+
+ private RemoteLogIndexCache remoteLogIndexCache(File dataDir) {
+ RemoteLogIndexCache remoteLogIndexCache = remoteLogIndexCachesByDir.get(dataDir);
+ if (remoteLogIndexCache == null) {
+ throw new IllegalArgumentException("Unknown local data directory: " + dataDir);
+ }
+ return remoteLogIndexCache;
+ }
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index 6a84316c3..1f8b893ab 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -230,6 +230,7 @@ public final class Replica {
private MetricGroup lakeTieringMetricGroup;
public Replica(
+ File dataDir,
PhysicalTablePath physicalPath,
TableBucket tableBucket,
LogManager logManager,
@@ -277,7 +278,7 @@ public final class Replica {
// create a closeable registry for the replica
this.closeableRegistry = new CloseableRegistry();
- this.logTablet = createLog(lazyHighWatermarkCheckpoint);
+ this.logTablet = createLog(dataDir, lazyHighWatermarkCheckpoint);
this.logTablet.updateIsDataLakeEnabled(tableConfig.isDataLakeEnabled());
this.clock = clock;
this.remoteLogManager = remoteLogManager;
@@ -384,7 +385,7 @@ public final class Replica {
}
public Path getTabletParentDir() {
- return logManager.getTabletParentDir(physicalPath, tableBucket);
+ return logManager.getTabletParentDir(logTablet.getDataDir(), physicalPath, tableBucket);
}
public @Nullable KvTablet getKvTablet() {
@@ -776,7 +777,9 @@ public final class Replica {
physicalPath);
CompletedSnapshot completedSnapshot = optCompletedSnapshot.get();
// always create a new dir for the kv tablet
- File tabletDir = kvManager.createTabletDir(physicalPath, tableBucket);
+ File tabletDir =
+ kvManager.createTabletDir(
+ logTablet.getDataDir(), physicalPath, tableBucket);
// down the snapshot to target tablet dir
downloadKvSnapshots(completedSnapshot, tabletDir.toPath());
@@ -2147,10 +2150,11 @@ public final class Replica {
}
private LogTablet createLog(
- OffsetCheckpointFile.LazyOffsetCheckpoints lazyHighWatermarkCheckpoint)
+ File dataDir, OffsetCheckpointFile.LazyOffsetCheckpoints lazyHighWatermarkCheckpoint)
throws Exception {
LogTablet log =
logManager.getOrCreateLog(
+ dataDir,
physicalPath,
tableBucket,
tableConfig.getLogFormat(),
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index f2a01cbe9..71fd233b6 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -106,6 +106,7 @@ import org.apache.fluss.server.replica.delay.DelayedTableBucketKey;
import org.apache.fluss.server.replica.delay.DelayedWrite;
import org.apache.fluss.server.replica.fetcher.InitialFetchStatus;
import org.apache.fluss.server.replica.fetcher.ReplicaFetcherManager;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.utils.FatalErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
@@ -161,7 +162,8 @@ public class ReplicaManager implements ServerReconfigurable {
private final ZooKeeperClient zkClient;
protected final int serverId;
private final AtomicBoolean highWatermarkCheckPointThreadStarted = new AtomicBoolean(false);
- private final OffsetCheckpointFile highWatermarkCheckpoint;
+ private final Map<File, OffsetCheckpointFile> highWatermarkCheckpoints;
+ private final LocalDiskManager localDiskManager;
@GuardedBy("replicaStateChangeLock")
private final Map<TableBucket, HostedReplica> allReplicas = new ConcurrentHashMap<>();
@@ -228,7 +230,8 @@ public class ReplicaManager implements ServerReconfigurable {
UserMetrics userMetrics,
ScannerManager scannerManager,
Clock clock,
- ExecutorService ioExecutor)
+ ExecutorService ioExecutor,
+ LocalDiskManager localDiskManager)
throws IOException {
this(
conf,
@@ -244,10 +247,18 @@ public class ReplicaManager implements ServerReconfigurable {
fatalErrorHandler,
serverMetricGroup,
userMetrics,
- new RemoteLogManager(conf, zkClient, coordinatorGateway, clock, ioExecutor),
+ new RemoteLogManager(
+ conf,
+ zkClient,
+ coordinatorGateway,
+ localDiskManager,
+ logManager,
+ clock,
+ ioExecutor),
scannerManager,
clock,
- ioExecutor);
+ ioExecutor,
+ localDiskManager);
}
@VisibleForTesting
@@ -268,21 +279,25 @@ public class ReplicaManager implements ServerReconfigurable {
RemoteLogManager remoteLogManager,
ScannerManager scannerManager,
Clock clock,
- ExecutorService ioExecutor)
+ ExecutorService ioExecutor,
+ LocalDiskManager localDiskManager)
throws IOException {
this.conf = conf;
this.zkClient = zkClient;
this.scheduler = scheduler;
+ this.localDiskManager = localDiskManager;
this.logManager = logManager;
this.kvManager = kvManager;
this.serverId = serverId;
this.metadataCache = metadataCache;
- this.highWatermarkCheckpoint =
- new OffsetCheckpointFile(
- new File(
- logManager.getDataDir().getAbsolutePath(),
- HIGH_WATERMARK_CHECKPOINT_FILE_NAME));
+ this.highWatermarkCheckpoints = new HashMap<>();
+ for (File dataDir : localDiskManager.dataDirs()) {
+ highWatermarkCheckpoints.put(
+ dataDir,
+ new OffsetCheckpointFile(
+ new File(dataDir, HIGH_WATERMARK_CHECKPOINT_FILE_NAME)));
+ }
this.delayedWriteManager =
new DelayedOperationManager<>(
"delay write",
@@ -1496,19 +1511,22 @@ public class ReplicaManager implements ServerReconfigurable {
// of RemoteLogSegment. For client fetcher, it will fetch the log from remote in client.
// For follower, it can update its local metadata to adjust the next fetch offset.
else if (canFetchFromRemoteLog(replica, fetchOffset)) {
- RemoteLogFetchInfo remoteLogFetchInfo = fetchLogFromRemote(replica, fetchOffset);
- if (remoteLogFetchInfo != null) {
- return new FetchLogResultForBucket(
- tb, remoteLogFetchInfo, replica.getLogHighWatermark());
- } else {
- return new FetchLogResultForBucket(
- tb,
- ApiError.fromThrowable(
- new LogOffsetOutOfRangeException(
- String.format(
- "The fetch offset %s is out of range for table bucket %s",
- fetchOffset, tb))));
+ try {
+ RemoteLogFetchInfo remoteLogFetchInfo = fetchLogFromRemote(replica, fetchOffset);
+ if (remoteLogFetchInfo != null) {
+ return new FetchLogResultForBucket(
+ tb, remoteLogFetchInfo, replica.getLogHighWatermark());
+ }
+ } catch (Exception ex) {
+ return new FetchLogResultForBucket(tb, ApiError.fromThrowable(ex));
}
+ return new FetchLogResultForBucket(
+ tb,
+ ApiError.fromThrowable(
+ new LogOffsetOutOfRangeException(
+ String.format(
+ "The fetch offset %s is out of range for table bucket %s",
+ fetchOffset, tb))));
} else {
return new FetchLogResultForBucket(tb, ApiError.fromThrowable(e));
}
@@ -1576,17 +1594,25 @@ public class ReplicaManager implements ServerReconfigurable {
@VisibleForTesting
void checkpointHighWatermarks() {
List<Replica> onlineReplicasList = getOnlineReplicaList();
- Map<TableBucket, Long> highWatermarks = new HashMap<>();
+ if (onlineReplicasList.isEmpty()) {
+ return;
+ }
+
+ Map<File, Map<TableBucket, Long>> highWatermarksByDir = new HashMap<>();
for (Replica replica : onlineReplicasList) {
LogTablet logTablet = replica.getLogTablet();
- highWatermarks.put(logTablet.getTableBucket(), logTablet.getHighWatermark());
+ highWatermarksByDir
+ .computeIfAbsent(logTablet.getDataDir(), ignored -> new HashMap<>())
+ .put(logTablet.getTableBucket(), logTablet.getHighWatermark());
}
- if (!highWatermarks.isEmpty()) {
- try {
- highWatermarkCheckpoint.write(highWatermarks);
- } catch (Exception e) {
- throw new LogStorageException("Error while writing to high watermark file", e);
+ for (Map.Entry<File, Map<TableBucket, Long>> entry : highWatermarksByDir.entrySet()) {
+ if (!entry.getValue().isEmpty()) {
+ try {
+ highWatermarkCheckpoints.get(entry.getKey()).write(entry.getValue());
+ } catch (Exception e) {
+ throw new LogStorageException("Error while writing to high watermark file", e);
+ }
}
}
}
@@ -1851,6 +1877,9 @@ public class ReplicaManager implements ServerReconfigurable {
serverMetricGroup.removeTableBucketMetricGroup(
replicaToDelete.getPhysicalTablePath().getTablePath(), tb);
replicaToDelete.delete();
+ localDiskManager.recordReplicaDelete(
+ replicaToDelete.getLogTablet().getDataDir(),
+ replicaToDelete.isKvTable());
Path tabletParentDir = replicaToDelete.getTabletParentDir();
if (tb.getPartitionId() != null) {
deletedPartitionIds.put(tb.getPartitionId(), tabletParentDir);
@@ -1932,11 +1961,20 @@ public class ReplicaManager implements ServerReconfigurable {
TableInfo tableInfo = getTableInfo(zkClient, tablePath);
boolean isKvTable = tableInfo.hasPrimaryKey();
+ Optional<LogTablet> existingLogTabletOpt = logManager.getLog(tb);
+ File dataDir =
+ existingLogTabletOpt
+ .map(LogTablet::getDataDir)
+ .orElseGet(
+ () ->
+ localDiskManager.selectDataDirForNewBucket(
+ isKvTable));
BucketMetricGroup bucketMetricGroup =
serverMetricGroup.addTableBucketMetricGroup(
physicalTablePath, tb, isKvTable);
Replica replica =
new Replica(
+ dataDir,
physicalTablePath,
tb,
logManager,
@@ -1945,7 +1983,7 @@ public class ReplicaManager implements ServerReconfigurable {
this::getMinInSyncReplicas,
serverId,
new OffsetCheckpointFile.LazyOffsetCheckpoints(
- highWatermarkCheckpoint),
+ highWatermarkCheckpoints.get(dataDir)),
delayedWriteManager,
delayedFetchLogManager,
adjustIsrManager,
@@ -1957,6 +1995,9 @@ public class ReplicaManager implements ServerReconfigurable {
clock,
remoteLogManager,
scannerManager);
+ if (!existingLogTabletOpt.isPresent()) {
+ localDiskManager.recordReplicaLoad(dataDir, isKvTable);
+ }
allReplicas.put(tb, new OnlineReplica(replica));
replicaOpt = Optional.of(replica);
} else if (hostedReplica instanceof OnlineReplica) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java b/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java
new file mode 100644
index 000000000..f7a8d6c44
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java
@@ -0,0 +1,747 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.storage;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.IllegalConfigurationException;
+import org.apache.fluss.exception.LogStorageException;
+import org.apache.fluss.utils.FileUtils;
+import org.apache.fluss.utils.IOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.channels.OverlappingFileLockException;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * Manages the local data directories used by a TabletServer.
+ *
+ * <p>The manager has two responsibilities:
+ *
+ * <ol>
+ * <li>During startup, validate configured data directories, recover or create {@code
+ * disk.properties}, and hold a {@code .lock} file for every live directory.
+ * <li>After startup, provide runtime helpers for path resolution and simple per-disk load
+ * accounting used by replica placement.
+ * </ol>
+ *
+ * <p>Thread-safety: initialization happens before the manager is published to other components.
+ * After that, the mutable state is limited to in-memory bucket counters and shutdown cleanup. The
+ * public runtime APIs synchronize on the manager instance so directory lookups, counter updates,
+ * and {@link #close()} observe a consistent view.
+ */
+public final class LocalDiskManager implements Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(LocalDiskManager.class);
+
+ public static final String DISK_PROPERTIES_FILE_NAME = "disk.properties";
+ public static final String LOCK_FILE_NAME = ".lock";
+ private static final int DISK_PROPERTIES_VERSION = 1;
+ private static final String VERSION_KEY = "version";
+ private static final String DISK_ID_KEY = "disk.id";
+ private static final String SERVER_ID_KEY = "server.id";
+
+ private final int serverId;
+ private final List<File> dataDirs;
+ private final Map<File, DiskInfo> diskInfosByDir = new LinkedHashMap<>();
+ private final Map<File, DirectoryLock> directoryLocks = new LinkedHashMap<>();
+
+ // ------------------------------------------------------------------------
+ // Initialization
+ // ------------------------------------------------------------------------
+
+ private LocalDiskManager(Configuration conf) throws IOException {
+ this.serverId = conf.getInt(ConfigOptions.TABLET_SERVER_ID);
+ List<File> configuredDataDirs = resolveConfiguredDataDirs(conf);
+ List<File> initializedDataDirs;
+ try {
+ initializedDataDirs = initializeDataDirs(configuredDataDirs);
+ } catch (Throwable throwable) {
+ close();
+ throw throwable;
+ }
+ this.dataDirs = Collections.unmodifiableList(initializedDataDirs);
+ }
+
+ public static LocalDiskManager create(Configuration conf) throws IOException {
+ LocalDiskManager localDiskManager = new LocalDiskManager(conf);
+ LOG.info(
+ "Initialized LocalDiskManager with local data directories {}.",
+ localDiskManager.dataDirs);
+ return localDiskManager;
+ }
+
+ /**
+ * Resolves {@code data.dirs} or falls back to {@code data.dir}, then normalizes every path
+ * exactly once.
+ *
+ * <p>The method rejects:
+ *
+ * <ul>
+ * <li>empty values, for example {@code ["/data-0", " "]}
+ * <li>duplicate directories, for example {@code ["/data-0", "/data-0"]}
+ * <li>overlapping directories, for example {@code ["/data", "/data/table-1"]}
+ * </ul>
+ *
+ * <p>Overlapping directories are rejected because they do not describe independent storage
+ * roots and would make placement and cleanup ambiguous.
+ */
+ private static List<File> resolveConfiguredDataDirs(Configuration conf) {
+ List<String> configuredDataDirs = conf.get(ConfigOptions.DATA_DIRS);
+ if (configuredDataDirs == null || configuredDataDirs.isEmpty()) {
+ configuredDataDirs = Collections.singletonList(conf.get(ConfigOptions.DATA_DIR));
+ }
+
+ List<File> dataDirs = new ArrayList<>(configuredDataDirs.size());
+ for (String configuredDataDir : configuredDataDirs) {
+ if (configuredDataDir == null || configuredDataDir.trim().isEmpty()) {
+ throw new IllegalConfigurationException(
+ "Configured local data directories must not contain empty values.");
+ }
+ File dataDir =
+ new File(configuredDataDir).toPath().toAbsolutePath().normalize().toFile();
+ for (File existingDataDir : dataDirs) {
+ if (existingDataDir.equals(dataDir)) {
+ throw new IllegalConfigurationException(
+ "Duplicate local data directory configured: " + dataDir);
+ }
+
+ Path existingPath = existingDataDir.toPath();
+ Path candidatePath = dataDir.toPath();
+ if (existingPath.startsWith(candidatePath)
+ || candidatePath.startsWith(existingPath)) {
+ throw new IllegalConfigurationException(
+ "Configured local data directories must not overlap: "
+ + existingPath
+ + " and "
+ + candidatePath);
+ }
+ }
+ dataDirs.add(dataDir);
+ }
+ return dataDirs;
+ }
+
+ /**
+ * Initializes all configured data directories and returns the directories that remain online
+ * after validation.
+ *
+ * <p>The initialization pipeline keeps dropping bad directories from {@code onlineDataDirs}. A
+ * directory can be removed at any step if it fails metadata loading, local directory
+ * validation, locking, or checkpoint writing. The remaining directories become the runtime
+ * storage roots.
+ */
+ private List<File> initializeDataDirs(List<File> configuredDataDirs) throws IOException {
+ Set<File> onlineDataDirs = new LinkedHashSet<>(configuredDataDirs);
+ Set<File> offlineDataDirs = new LinkedHashSet<>();
+
+ // 1. Create disk.properties checkpoints for all configured data directories.
+ Map<File, DiskPropertiesCheckpoint> diskPropertiesCheckpoints =
+ newDiskPropertiesCheckpoints(configuredDataDirs);
+ // 2. Load persisted disk metadata from disk.properties.
+ Map<File, DiskProperties> diskPropertiesMap =
+ loadExistingDiskProperties(
+ onlineDataDirs, offlineDataDirs, diskPropertiesCheckpoints);
+ // 3. Create and validate the configured data directories.
+ createAndValidateDataDirs(onlineDataDirs, offlineDataDirs);
+ // 4. Acquire .lock for all directories that are still online.
+ lockDataDirs(onlineDataDirs, offlineDataDirs);
+ // 5. Assign disk ids and write disk.properties for directories that do not have it yet.
+ initializeMissingDiskProperties(
+ onlineDataDirs, offlineDataDirs, diskPropertiesMap, diskPropertiesCheckpoints);
+ // 6. Fail fast if every configured directory has been dropped during initialization.
+ validateNonEmptyDataDirs(configuredDataDirs, onlineDataDirs);
+ // 7. Publish the runtime disk state used by placement and path resolution.
+ finishDiskInfos(onlineDataDirs, diskPropertiesMap);
+
+ return new ArrayList<>(onlineDataDirs);
+ }
+
+ private static Map<File, DiskPropertiesCheckpoint> newDiskPropertiesCheckpoints(
+ List<File> configuredDataDirs) {
+ Map<File, DiskPropertiesCheckpoint> checkpoints = new LinkedHashMap<>();
+ for (File dataDir : configuredDataDirs) {
+ checkpoints.put(
+ dataDir,
+ new DiskPropertiesCheckpoint(new File(dataDir, DISK_PROPERTIES_FILE_NAME)));
+ }
+ return checkpoints;
+ }
+
+ private Map<File, DiskProperties> loadExistingDiskProperties(
+ Set<File> onlineDataDirs,
+ Set<File> offlineDataDirs,
+ Map<File, DiskPropertiesCheckpoint> diskPropertiesCheckpoints) {
+ Map<File, DiskProperties> diskPropertiesMap = new LinkedHashMap<>();
+ Set<Integer> persistedServerIdSet = new LinkedHashSet<>();
+ Map<String, File> diskIdToDataDir = new LinkedHashMap<>();
+
+ for (File dataDir : new ArrayList<>(onlineDataDirs)) {
+ try {
+ DiskProperties diskProperties = diskPropertiesCheckpoints.get(dataDir).read();
+ if (diskProperties != null) {
+ int persistedServerId = diskProperties.serverIdAsInt(dataDir);
+ String persistedDiskId = diskProperties.diskId(dataDir);
+ diskPropertiesMap.put(dataDir, diskProperties);
+ persistedServerIdSet.add(persistedServerId);
+
+ File previousDataDir = diskIdToDataDir.putIfAbsent(persistedDiskId, dataDir);
+ if (previousDataDir != null) {
+ throw new IllegalConfigurationException(
+ "Configured local data directories contain duplicate disk.id "
+ + persistedDiskId
+ + " for "
+ + previousDataDir
+ + " and "
+ + dataDir);
+ }
+ }
+ } catch (IOException e) {
+ markDataDirOffline(
+ dataDir,
+ onlineDataDirs,
+ offlineDataDirs,
+ "Failed to read disk.properties for data directory " + dataDir,
+ e);
+ }
+ }
+
+ if (persistedServerIdSet.size() > 1) {
+ StringBuilder builder = new StringBuilder();
+ for (Map.Entry<File, DiskProperties> entry : diskPropertiesMap.entrySet()) {
+ builder.append("- ")
+ .append(entry.getKey())
+ .append(" -> ")
+ .append(entry.getValue())
+ .append('\n');
+ }
+ throw new IllegalConfigurationException(
+ "disk.properties is not consistent across data.dirs. This could happen if "
+ + "multiple tablet servers shared a data directory (data.dirs) or partial data was "
+ + "manually copied from another server. Found:\n"
+ + builder);
+ } else if (persistedServerIdSet.size() == 1 && !persistedServerIdSet.contains(serverId)) {
+ int persistedServerId = persistedServerIdSet.iterator().next();
+ throw new IllegalConfigurationException(
+ "Configured tablet server id "
+ + serverId
+ + " does not match server.id "
+ + persistedServerId
+ + " in disk.properties. If you moved your data, make sure the "
+ + "configured tablet server id matches. If you intend to create a "
+ + "new tablet server, remove all data under your data directories "
+ + "(data.dirs).");
+ }
+
+ return diskPropertiesMap;
+ }
+
+ private void createAndValidateDataDirs(Set<File> onlineDataDirs, Set<File> offlineDataDirs) {
+ for (File dataDir : new ArrayList<>(onlineDataDirs)) {
+ try {
+ if (!dataDir.exists()) {
+ LOG.info(
+ "Data directory {} not found, creating it.", dataDir.getAbsolutePath());
+ if (!dataDir.mkdirs()) {
+ throw new IOException(
+ "Failed to create data directory " + dataDir.getAbsolutePath());
+ }
+ Path parentPath = dataDir.toPath().toAbsolutePath().normalize().getParent();
+ FileUtils.flushDir(parentPath);
+ }
+ if (!dataDir.isDirectory() || !dataDir.canRead()) {
+ throw new IOException(
+ dataDir.getAbsolutePath() + " is not a readable data directory.");
+ }
+ } catch (IOException e) {
+ markDataDirOffline(
+ dataDir,
+ onlineDataDirs,
+ offlineDataDirs,
+ "Failed to create or validate data directory " + dataDir,
+ e);
+ }
+ }
+ }
+
+ private void lockDataDirs(Set<File> onlineDataDirs, Set<File> offlineDataDirs) {
+ for (File dataDir : new ArrayList<>(onlineDataDirs)) {
+ try {
+ DirectoryLock directoryLock = new DirectoryLock(new File(dataDir, LOCK_FILE_NAME));
+ if (!directoryLock.tryLock()) {
+ closeDirectoryLock(directoryLock);
+ // Only real disk/I/O failures should be downgraded to offline data dirs.
+ // A lock conflict means another Fluss process/thread already owns this
+ // directory, so startup should fail fast instead of silently skipping it.
+ throw new IllegalConfigurationException(
+ "Failed to acquire lock on file "
+ + LOCK_FILE_NAME
+ + " in "
+ + directoryLock.file.getParent()
+ + ". A Fluss tablet server instance in another process or "
+ + "thread is using this directory.");
+ }
+ directoryLocks.put(dataDir, directoryLock);
+ } catch (IOException e) {
+ markDataDirOffline(
+ dataDir,
+ onlineDataDirs,
+ offlineDataDirs,
+ "Failed to acquire lock for data directory " + dataDir,
+ e);
+ }
+ }
+ }
+
+ private void initializeMissingDiskProperties(
+ Set<File> onlineDataDirs,
+ Set<File> offlineDataDirs,
+ Map<File, DiskProperties> diskPropertiesMap,
+ Map<File, DiskPropertiesCheckpoint> diskPropertiesCheckpoints) {
+
+ // 1. Find directories that are live but do not have persisted local metadata yet.
+ Set<File> dataDirsWithoutDiskProperties = new LinkedHashSet<>();
+ for (File dataDir : onlineDataDirs) {
+ if (!diskPropertiesMap.containsKey(dataDir)) {
+ dataDirsWithoutDiskProperties.add(dataDir);
+ }
+ }
+ // 2. Collect disk ids that are already persisted so we do not assign duplicates locally.
+ Set<String> usedDiskIds = new LinkedHashSet<>();
+ for (DiskProperties diskProperties : diskPropertiesMap.values()) {
+ usedDiskIds.add(diskProperties.diskId);
+ }
+ // 3. Create the missing metadata in memory first.
+ for (File dataDir : dataDirsWithoutDiskProperties) {
+ diskPropertiesMap.put(
+ dataDir,
+ new DiskProperties(
+ DISK_PROPERTIES_VERSION,
+ newDiskId(usedDiskIds),
+ String.valueOf(serverId)));
+ }
+ // 4. Persist the missing metadata. A write failure makes the directory unusable for this
+ // startup, so the directory is unlocked and moved to offlineDataDirs.
+ for (File dataDir : new ArrayList<>(dataDirsWithoutDiskProperties)) {
+ DiskProperties diskProperties = diskPropertiesMap.get(dataDir);
+ try {
+ diskPropertiesCheckpoints.get(dataDir).write(diskProperties);
+ } catch (IOException e) {
+ closeDirectoryLock(dataDir);
+ diskPropertiesMap.remove(dataDir);
+ markDataDirOffline(
+ dataDir,
+ onlineDataDirs,
+ offlineDataDirs,
+ "Failed to checkpoint disk.properties for data directory " + dataDir,
+ e);
+ }
+ }
+ }
+
+ private void validateNonEmptyDataDirs(List<File> configuredDataDirs, Set<File> onlineDataDirs) {
+ if (onlineDataDirs.isEmpty()) {
+ throw new LogStorageException(
+ "None of the specified data dirs from "
+ + configuredDataDirs
+ + " can be created, locked or checkpointed");
+ }
+ }
+
+ private void finishDiskInfos(
+ Set<File> onlineDataDirs, Map<File, DiskProperties> diskPropertiesMap) {
+ for (File dataDir : onlineDataDirs) {
+ DiskProperties diskProperties = diskPropertiesMap.get(dataDir);
+ diskInfosByDir.put(dataDir, new DiskInfo(dataDir, diskProperties.diskId));
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Public APIs
+ // ------------------------------------------------------------------------
+
+ public List<File> dataDirs() {
+ return dataDirs;
+ }
+
+ /**
+ * Resolves a local path back to the configured data directory that owns it.
+ *
+ * <p>Example: if {@code /data-0} is configured and the path is {@code /data-0/db/tablet-1}, the
+ * returned directory is {@code /data-0}.
+ */
+ public synchronized File resolveDataDir(File path) {
+ Path pathToResolve = path.toPath().toAbsolutePath().normalize();
+ for (File dataDir : dataDirs) {
+ if (pathToResolve.startsWith(dataDir.toPath())) {
+ return dataDir;
+ }
+ }
+
+ throw new IllegalArgumentException(
+ "Path " + path + " does not belong to any configured data directory.");
+ }
+
+ /**
+ * Selects a target directory for a new bucket replica.
+ *
+ * <p>Log-table buckets use {@code logBucketCount}. Primary-key table buckets use {@code
+ * kvBucketCount} so KV-heavy workloads are balanced independently from pure log workloads.
+ */
+ public synchronized File selectDataDirForNewBucket(boolean hasPrimaryKey) {
+ DiskInfo selectedDisk = null;
+ Comparator<DiskInfo> comparator =
+ hasPrimaryKey
+ ? Comparator.comparingInt((DiskInfo diskInfo) -> diskInfo.kvBucketCount)
+ : Comparator.comparingInt((DiskInfo diskInfo) -> diskInfo.logBucketCount);
+ for (DiskInfo candidate : diskInfosByDir.values()) {
+ if (selectedDisk == null || comparator.compare(candidate, selectedDisk) < 0) {
+ selectedDisk = candidate;
+ }
+ }
+ if (selectedDisk == null) {
+ throw new IllegalStateException("No configured data directory is available.");
+ }
+ return selectedDisk.dataDir;
+ }
+
+ public synchronized void recordReplicaLoad(File dataDir, boolean hasPrimaryKey) {
+ DiskInfo diskInfo = diskInfo(dataDir);
+ diskInfo.logBucketCount++;
+ if (hasPrimaryKey) {
+ diskInfo.kvBucketCount++;
+ }
+ }
+
+ public synchronized void recordReplicaDelete(File dataDir, boolean hasPrimaryKey) {
+ DiskInfo diskInfo = diskInfo(dataDir);
+ diskInfo.logBucketCount = Math.max(0, diskInfo.logBucketCount - 1);
+ if (hasPrimaryKey) {
+ diskInfo.kvBucketCount = Math.max(0, diskInfo.kvBucketCount - 1);
+ }
+ }
+
+ @VisibleForTesting
+ synchronized String diskId(File dataDir) {
+ return diskInfo(dataDir).diskId;
+ }
+
+ @VisibleForTesting
+ synchronized int logBucketCount(File dataDir) {
+ return diskInfo(dataDir).logBucketCount;
+ }
+
+ @VisibleForTesting
+ synchronized int kvBucketCount(File dataDir) {
+ return diskInfo(dataDir).kvBucketCount;
+ }
+
+ @Override
+ public synchronized void close() {
+ for (DirectoryLock directoryLock : directoryLocks.values()) {
+ closeDirectoryLock(directoryLock);
+ }
+ directoryLocks.clear();
+ diskInfosByDir.clear();
+ }
+
+ // ------------------------------------------------------------------------
+ // Internal helpers
+ // ------------------------------------------------------------------------
+
+ private void markDataDirOffline(
+ File dataDir,
+ Set<File> onlineDataDirs,
+ Set<File> offlineDataDirs,
+ String message,
+ Exception exception) {
+ onlineDataDirs.remove(dataDir);
+ offlineDataDirs.add(dataDir);
+ if (exception == null) {
+ LOG.error(message);
+ } else {
+ LOG.error(message, exception);
+ }
+ }
+
+ private static String newDiskId(Set<String> usedDiskIds) {
+ String diskId;
+ do {
+ diskId = UUID.randomUUID().toString();
+ } while (!usedDiskIds.add(diskId));
+ return diskId;
+ }
+
+ private DiskInfo diskInfo(File dataDir) {
+ DiskInfo diskInfo = diskInfosByDir.get(dataDir);
+ if (diskInfo == null) {
+ throw new IllegalArgumentException("Unknown data directory: " + dataDir);
+ }
+ return diskInfo;
+ }
+
+ private void closeDirectoryLock(File dataDir) {
+ DirectoryLock directoryLock = directoryLocks.remove(dataDir);
+ if (directoryLock != null) {
+ closeDirectoryLock(directoryLock);
+ }
+ }
+
+ private void closeDirectoryLock(DirectoryLock directoryLock) {
+ try {
+ directoryLock.destroy();
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to close directory lock for {}", directoryLock.file.getParentFile(), e);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Core data objects
+ // ------------------------------------------------------------------------
+
+ /** Parsed contents of a {@code disk.properties} file. */
+ private static final class DiskProperties {
+ private final int version;
+ private final String diskId;
+ private final String serverId;
+
+ private DiskProperties(int version, String diskId, String serverId) {
+ this.version = version;
+ this.diskId = diskId == null ? null : diskId.trim();
+ this.serverId = serverId == null ? null : serverId.trim();
+ }
+
+ private int serverIdAsInt(File dataDir) throws IOException {
+ if (serverId == null || serverId.isEmpty()) {
+ throw new IOException(
+ "Missing server.id in disk.properties under " + dataDir.getAbsolutePath());
+ }
+
+ try {
+ return Integer.parseInt(serverId);
+ } catch (NumberFormatException e) {
+ throw new IOException(
+ "Invalid server.id in disk.properties under "
+ + dataDir.getAbsolutePath()
+ + ": "
+ + serverId,
+ e);
+ }
+ }
+
+ private String diskId(File dataDir) throws IOException {
+ if (diskId == null || diskId.isEmpty()) {
+ throw new IOException(
+ "Missing disk.id in disk.properties under " + dataDir.getAbsolutePath());
+ }
+ return diskId;
+ }
+
+ @Override
+ public String toString() {
+ return "DiskProperties(version="
+ + version
+ + ", diskId="
+ + diskId
+ + ", serverId="
+ + serverId
+ + ")";
+ }
+ }
+
+ /** Read/write helper for a single {@code disk.properties} file. */
+ private static final class DiskPropertiesCheckpoint {
+ private final File file;
+ private final Object lock = new Object();
+
+ private DiskPropertiesCheckpoint(File file) {
+ this.file = file;
+ }
+
+ private void write(DiskProperties diskProperties) throws IOException {
+ synchronized (lock) {
+ try {
+ File tempFile = new File(file.getAbsolutePath() + ".tmp");
+ FileOutputStream outputStream = new FileOutputStream(tempFile);
+ try {
+ Properties properties = new Properties();
+ properties.setProperty(VERSION_KEY, String.valueOf(diskProperties.version));
+ properties.setProperty(DISK_ID_KEY, diskProperties.diskId);
+ properties.setProperty(SERVER_ID_KEY, diskProperties.serverId);
+ properties.store(outputStream, "");
+ outputStream.flush();
+ outputStream.getFD().sync();
+ } finally {
+ IOUtils.closeQuietly(outputStream, tempFile.getName());
+ }
+ FileUtils.atomicMoveWithFallback(tempFile.toPath(), file.toPath(), false);
+ } catch (IOException e) {
+ LOG.error("Failed to write disk.properties due to", e);
+ throw e;
+ }
+ }
+ }
+
+ private DiskProperties read() throws IOException {
+ Files.deleteIfExists(new File(file.getPath() + ".tmp").toPath());
+
+ synchronized (lock) {
+ InputStream inputStream = null;
+ try {
+ inputStream = Files.newInputStream(file.toPath());
+ Properties properties = new Properties();
+ properties.load(inputStream);
+
+ String version = properties.getProperty(VERSION_KEY);
+ if (version == null) {
+ throw new IOException(
+ "Unrecognized version of disk.properties file under "
+ + file.getAbsolutePath()
+ + ": null");
+ }
+
+ int parsedVersion;
+ try {
+ parsedVersion = Integer.parseInt(version);
+ } catch (NumberFormatException e) {
+ throw new IOException(
+ "Unrecognized version of disk.properties file under "
+ + file.getAbsolutePath()
+ + ": "
+ + version,
+ e);
+ }
+
+ if (parsedVersion != DISK_PROPERTIES_VERSION) {
+ throw new IOException(
+ "Unrecognized version of disk.properties file under "
+ + file.getAbsolutePath()
+ + ": "
+ + version);
+ }
+
+ String diskId = properties.getProperty(DISK_ID_KEY);
+ String serverId = properties.getProperty(SERVER_ID_KEY);
+ return new DiskProperties(parsedVersion, diskId, serverId);
+ } catch (NoSuchFileException e) {
+ LOG.info("No disk.properties file under dir {}", file.getAbsolutePath());
+ return null;
+ } catch (Exception e) {
+ LOG.error(
+ "Failed to read disk.properties file under dir {} due to {}",
+ file.getAbsolutePath(),
+ e.getMessage());
+ if (e instanceof IOException) {
+ throw (IOException) e;
+ }
+ throw new IOException(
+ "Failed to read disk.properties file under dir "
+ + file.getAbsolutePath(),
+ e);
+ } finally {
+ IOUtils.closeQuietly(inputStream, file.getName());
+ }
+ }
+ }
+ }
+
+ /** Holds the {@code .lock} file for one live data directory. */
+ private static final class DirectoryLock {
+ private final File file;
+ private final FileChannel channel;
+ private FileLock flock;
+
+ private DirectoryLock(File file) throws IOException {
+ this(
+ file,
+ FileChannel.open(
+ file.toPath(),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE),
+ null);
+ }
+
+ private DirectoryLock(File file, FileChannel channel, FileLock flock) {
+ this.file = file;
+ this.channel = channel;
+ this.flock = flock;
+ }
+
+ private synchronized void lock() throws IOException {
+ flock = channel.lock();
+ }
+
+ private synchronized boolean tryLock() throws IOException {
+ try {
+ flock = channel.tryLock();
+ return flock != null;
+ } catch (OverlappingFileLockException e) {
+ return false;
+ }
+ }
+
+ private synchronized void unlock() throws IOException {
+ if (flock != null) {
+ flock.release();
+ }
+ }
+
+ private synchronized void destroy() throws IOException {
+ unlock();
+ channel.close();
+ }
+ }
+
+ /** Runtime state for one online data directory. */
+ private static final class DiskInfo {
+ private final File dataDir;
+ private final String diskId;
+ private int logBucketCount;
+ private int kvBucketCount;
+
+ private DiskInfo(File dataDir, String diskId) {
+ this.dataDir = dataDir;
+ this.diskId = diskId;
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index d0f57db64..4ec7df46e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -49,6 +49,7 @@ import org.apache.fluss.server.metrics.ServerMetricUtils;
import org.apache.fluss.server.metrics.UserMetrics;
import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
import org.apache.fluss.server.replica.ReplicaManager;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperUtils;
import org.apache.fluss.server.zk.data.TabletServerRegistration;
@@ -143,6 +144,9 @@ public class TabletServer extends ServerBase {
@GuardedBy("lock")
private KvManager kvManager;
+ @GuardedBy("lock")
+ private LocalDiskManager localDiskManager;
+
@GuardedBy("lock")
private ReplicaManager replicaManager;
@@ -227,11 +231,20 @@ public class TabletServer extends ServerBase {
this.metadataCache = new TabletServerMetadataCache(metadataManager);
+ this.localDiskManager = LocalDiskManager.create(conf);
this.logManager =
- LogManager.create(conf, zkClient, scheduler, clock, tabletServerMetricGroup);
+ LogManager.create(
+ conf,
+ zkClient,
+ scheduler,
+ clock,
+ tabletServerMetricGroup,
+ localDiskManager);
logManager.startup();
- this.kvManager = KvManager.create(conf, zkClient, logManager, tabletServerMetricGroup);
+ this.kvManager =
+ KvManager.create(
+ conf, zkClient, logManager, tabletServerMetricGroup, localDiskManager);
kvManager.startup();
// Register kvManager to dynamicConfigManager for dynamic reconfiguration
@@ -278,7 +291,8 @@ public class TabletServer extends ServerBase {
userMetrics,
scannerManager,
clock,
- ioExecutor);
+ ioExecutor,
+ localDiskManager);
replicaManager.startup();
// Register DefaultSnapshotContext for dynamic kv.snapshot.interval
@@ -461,6 +475,10 @@ public class TabletServer extends ServerBase {
replicaManager.shutdown();
}
+ if (localDiskManager != null) {
+ localDiskManager.close();
+ }
+
if (authorizer != null) {
authorizer.close();
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java
index cecff0f8f..2c54c1665 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java
@@ -36,6 +36,7 @@ import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.server.log.LogManager;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
@@ -91,6 +92,7 @@ final class KvManagerTest {
private TableBucket tableBucket1;
private TableBucket tableBucket2;
+ private LocalDiskManager localDiskManager;
private LogManager logManager;
private KvManager kvManager;
private Configuration conf;
@@ -107,34 +109,43 @@ final class KvManagerTest {
void setup() throws Exception {
conf = new Configuration();
conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath());
+ conf.set(ConfigOptions.TABLET_SERVER_ID, 1);
String dbName = "db1";
tablePath1 = TablePath.of(dbName, "t1");
tablePath2 = TablePath.of(dbName, "t2");
// we need a log manager for kv manager
-
+ localDiskManager = LocalDiskManager.create(conf);
logManager =
LogManager.create(
conf,
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
kvManager =
KvManager.create(
- conf, zkClient, logManager, TestingMetricGroups.TABLET_SERVER_METRICS);
+ conf,
+ zkClient,
+ logManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
kvManager.startup();
}
@AfterEach
- void tearDown() {
+ void tearDown() throws Exception {
if (kvManager != null) {
kvManager.shutdown();
}
if (logManager != null) {
logManager.shutdown();
}
+ if (localDiskManager != null) {
+ localDiskManager.close();
+ }
}
static List<String> partitionProvider() {
@@ -185,7 +196,11 @@ final class KvManagerTest {
kvManager.shutdown();
kvManager =
KvManager.create(
- conf, zkClient, logManager, TestingMetricGroups.TABLET_SERVER_METRICS);
+ conf,
+ zkClient,
+ logManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
kvManager.startup();
kv1 = getOrCreateKv(tablePath1, partitionName, tableBucket1);
kv2 = getOrCreateKv(tablePath2, partitionName, tableBucket2);
@@ -230,7 +245,11 @@ final class KvManagerTest {
testingSchemaGetter.updateLatestSchemaInfo(new SchemaInfo(DATA2_SCHEMA, newSchemaId));
kvManager =
KvManager.create(
- conf, zkClient, logManager, TestingMetricGroups.TABLET_SERVER_METRICS);
+ conf,
+ zkClient,
+ logManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
kvManager.startup();
// insert again after restart.
@@ -351,7 +370,8 @@ final class KvManagerTest {
PhysicalTablePath.of(
tablePath.getDatabaseName(), tablePath.getTableName(), partitionName);
LogTablet logTablet =
- logManager.getOrCreateLog(physicalTablePath, tableBucket, LogFormat.ARROW, 1, true);
+ logManager.getOrCreateLog(
+ tempDir, physicalTablePath, tableBucket, LogFormat.ARROW, 1, true);
return kvManager.getOrCreateKv(
physicalTablePath,
tableBucket,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java
index 650a23a9c..97ae1d9cc 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java
@@ -121,6 +121,7 @@ class KvTabletMergeModeTest {
physicalTablePath.getTableName());
logTablet =
LogTablet.create(
+ tempLogDir,
physicalTablePath,
logTabletDir,
conf,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java
index d5ed5b30f..130352170 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java
@@ -109,6 +109,7 @@ class KvTabletSchemaEvolutionTest {
physicalTablePath.getTableName());
logTablet =
LogTablet.create(
+ tempLogDir,
physicalTablePath,
logTabletDir,
conf,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
index 1f2d5a89f..d7ef34701 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -170,6 +170,7 @@ class KvTabletTest {
LogTestUtils.makeRandomLogTabletDir(
tempLogDir, tablePath.getDatabaseName(), tableId, tablePath.getTableName());
return LogTablet.create(
+ tempLogDir,
tablePath,
logTabletDir,
conf,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java
index c04a907b3..3843290f5 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/scan/ScannerManagerTest.java
@@ -99,6 +99,7 @@ class ScannerManagerTest {
physicalTablePath.getTableName());
logTablet =
LogTablet.create(
+ tempLogDir,
physicalTablePath,
logTabletDir,
conf,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/DroppedTableRecoveryTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/DroppedTableRecoveryTest.java
index 021e9d1b0..e3d5b9ab3 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/DroppedTableRecoveryTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/DroppedTableRecoveryTest.java
@@ -32,6 +32,7 @@ import org.apache.fluss.record.TestingSchemaGetter;
import org.apache.fluss.server.kv.KvManager;
import org.apache.fluss.server.kv.KvTablet;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
@@ -70,6 +71,7 @@ final class DroppedTableRecoveryTest extends LogTestBase {
private @TempDir File tempDir;
private TablePath tablePath;
private TableBucket tableBucket;
+ private LocalDiskManager localDiskManager;
private LogManager logManager;
private KvManager kvManager;
@@ -85,24 +87,31 @@ final class DroppedTableRecoveryTest extends LogTestBase {
public void setup() throws Exception {
super.before();
conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath());
+ conf.set(ConfigOptions.TABLET_SERVER_ID, 1);
String dbName = "test_db";
tablePath = TablePath.of(dbName, "dropped_table");
tableBucket = new TableBucket(DATA1_TABLE_ID, 1);
registerTableInZkClient();
+ localDiskManager = LocalDiskManager.create(conf);
logManager =
LogManager.create(
conf,
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
logManager.startup();
kvManager =
KvManager.create(
- conf, zkClient, logManager, TestingMetricGroups.TABLET_SERVER_METRICS);
+ conf,
+ zkClient,
+ logManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
kvManager.startup();
}
@@ -116,13 +125,16 @@ final class DroppedTableRecoveryTest extends LogTestBase {
}
@AfterEach
- public void tearDown() {
+ public void tearDown() throws Exception {
if (kvManager != null) {
kvManager.shutdown();
}
if (logManager != null) {
logManager.shutdown();
}
+ if (localDiskManager != null) {
+ localDiskManager.close();
+ }
}
@Test
@@ -133,10 +145,20 @@ final class DroppedTableRecoveryTest extends LogTestBase {
LogTablet log1 =
logManager.getOrCreateLog(
- PhysicalTablePath.of(tablePath), tableBucket1, LogFormat.ARROW, 1, false);
+ tempDir,
+ PhysicalTablePath.of(tablePath),
+ tableBucket1,
+ LogFormat.ARROW,
+ 1,
+ false);
LogTablet log2 =
logManager.getOrCreateLog(
- PhysicalTablePath.of(tablePath), tableBucket2, LogFormat.ARROW, 1, false);
+ tempDir,
+ PhysicalTablePath.of(tablePath),
+ tableBucket2,
+ LogFormat.ARROW,
+ 1,
+ false);
// Write some data to both logs
MemoryLogRecords records = genMemoryLogRecordsByObject(DATA1);
@@ -151,18 +173,21 @@ final class DroppedTableRecoveryTest extends LogTestBase {
// Shutdown log manager first
logManager.shutdown();
+ localDiskManager.close();
// Simulate table drop: remove metadata from ZooKeeper
zkClient.deleteTable(tablePath);
// Start LogManager again
+ localDiskManager = LocalDiskManager.create(conf);
LogManager newLogManager =
LogManager.create(
conf,
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
newLogManager.startup();
// Verify that both residual data directories were cleaned up
@@ -181,7 +206,12 @@ final class DroppedTableRecoveryTest extends LogTestBase {
LogTablet log =
logManager.getOrCreateLog(
- partitionedTablePath, partitionedTableBucket, LogFormat.ARROW, 1, false);
+ tempDir,
+ partitionedTablePath,
+ partitionedTableBucket,
+ LogFormat.ARROW,
+ 1,
+ false);
// Write some data to the log
MemoryLogRecords records = genMemoryLogRecordsByObject(DATA1);
@@ -192,18 +222,21 @@ final class DroppedTableRecoveryTest extends LogTestBase {
// Shutdown log manager first
logManager.shutdown();
+ localDiskManager.close();
// Simulate table drop: remove metadata from ZooKeeper
zkClient.deleteTable(tablePath);
// Start LogManager again
+ localDiskManager = LocalDiskManager.create(conf);
LogManager newLogManager =
LogManager.create(
conf,
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
newLogManager.startup();
// Verify that the residual data directory was cleaned up
@@ -220,10 +253,20 @@ final class DroppedTableRecoveryTest extends LogTestBase {
LogTablet log1 =
logManager.getOrCreateLog(
- PhysicalTablePath.of(tablePath), tableBucket1, LogFormat.ARROW, 1, false);
+ tempDir,
+ PhysicalTablePath.of(tablePath),
+ tableBucket1,
+ LogFormat.ARROW,
+ 1,
+ false);
LogTablet log2 =
logManager.getOrCreateLog(
- PhysicalTablePath.of(tablePath), tableBucket2, LogFormat.ARROW, 1, false);
+ tempDir,
+ PhysicalTablePath.of(tablePath),
+ tableBucket2,
+ LogFormat.ARROW,
+ 1,
+ false);
// Write some data to both logs
MemoryLogRecords records = genMemoryLogRecordsByObject(DATA1);
@@ -263,23 +306,30 @@ final class DroppedTableRecoveryTest extends LogTestBase {
// Shutdown managers first
kvManager.shutdown();
logManager.shutdown();
+ localDiskManager.close();
// Simulate table drop: remove metadata from ZooKeeper
zkClient.deleteTable(tablePath);
// Start managers again
+ localDiskManager = LocalDiskManager.create(conf);
LogManager newLogManager =
LogManager.create(
conf,
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
newLogManager.startup(); // Should clean up log directories
KvManager newKvManager =
KvManager.create(
- conf, zkClient, newLogManager, TestingMetricGroups.TABLET_SERVER_METRICS);
+ conf,
+ zkClient,
+ newLogManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
newKvManager.startup();
// KV tablet directories should be cleaned up by LogManager automatically
@@ -303,7 +353,12 @@ final class DroppedTableRecoveryTest extends LogTestBase {
LogTablet log =
logManager.getOrCreateLog(
- partitionedTablePath, partitionedTableBucket, LogFormat.ARROW, 1, false);
+ tempDir,
+ partitionedTablePath,
+ partitionedTableBucket,
+ LogFormat.ARROW,
+ 1,
+ false);
// Write some data to the log
MemoryLogRecords records = genMemoryLogRecordsByObject(DATA1);
@@ -329,23 +384,30 @@ final class DroppedTableRecoveryTest extends LogTestBase {
// Shutdown managers first
kvManager.shutdown();
logManager.shutdown();
+ localDiskManager.close();
// Simulate table drop: remove metadata from ZooKeeper
zkClient.deleteTable(tablePath);
// Start managers again
+ localDiskManager = LocalDiskManager.create(conf);
LogManager newLogManager =
LogManager.create(
conf,
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
newLogManager.startup();
KvManager newKvManager =
KvManager.create(
- conf, zkClient, newLogManager, TestingMetricGroups.TABLET_SERVER_METRICS);
+ conf,
+ zkClient,
+ newLogManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
newKvManager.startup();
// KV tablet directory should be cleaned up by LogManager automatically
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
index 03d2ddadd..582b2934d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
@@ -634,6 +634,7 @@ final class LogLoaderTest extends LogTestBase {
private LogTablet createLogTablet(boolean isCleanShutdown, long recoveryPoint)
throws Exception {
return LogTablet.create(
+ tempDir,
PhysicalTablePath.of(DATA1_TABLE_PATH),
logDir,
conf,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
index 981685a3d..a6087e39b 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java
@@ -26,6 +26,8 @@ import org.apache.fluss.record.LogTestBase;
import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.server.log.checkpoint.OffsetCheckpointFile;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
+import org.apache.fluss.server.storage.LocalDiskManager;
+import org.apache.fluss.server.testutils.ServerTestTags;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
@@ -37,7 +39,9 @@ import org.apache.fluss.utils.concurrent.FlussScheduler;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
@@ -68,6 +72,7 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link LogManager}. */
final class LogManagerTest extends LogTestBase {
+
@RegisterExtension
public static final AllCallbackWrapper<ZooKeeperExtension> ZOO_KEEPER_EXTENSION_WRAPPER =
new AllCallbackWrapper<>(new ZooKeeperExtension());
@@ -78,6 +83,7 @@ final class LogManagerTest extends LogTestBase {
private TablePath tablePath2;
private TableBucket tableBucket1;
private TableBucket tableBucket2;
+ private LocalDiskManager localDiskManager;
private LogManager logManager;
// TODO add more tests refer to kafka's LogManagerTest.
@@ -91,23 +97,34 @@ final class LogManagerTest extends LogTestBase {
}
@BeforeEach
- public void setup() throws Exception {
+ public void setup(TestInfo testInfo) throws Exception {
super.before();
- conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath());
+ if (testInfo.getTags().contains(ServerTestTags.JBOD_MULTI_DIR_TAG)) {
+ conf.set(
+ ConfigOptions.DATA_DIRS,
+ Arrays.asList(
+ new File(tempDir, "data-1").getAbsolutePath(),
+ new File(tempDir, "data-2").getAbsolutePath()));
+ } else {
+ conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath());
+ }
conf.setString(ConfigOptions.COORDINATOR_HOST, "localhost");
+ conf.set(ConfigOptions.TABLET_SERVER_ID, 1);
String dbName = "db1";
tablePath1 = TablePath.of(dbName, "t1");
tablePath2 = TablePath.of(dbName, "t2");
registerTableInZkClient();
+ localDiskManager = LocalDiskManager.create(conf);
logManager =
LogManager.create(
conf,
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
logManager.startup();
}
@@ -176,7 +193,7 @@ final class LogManagerTest extends LogTestBase {
}
log2.flush(false);
- logManager.checkpointRecoveryOffsets();
+ logManager.checkpointRecoveryOffsets(tempDir);
Map<TableBucket, Long> checkpoints =
new OffsetCheckpointFile(
new File(tempDir, LogManager.RECOVERY_POINT_CHECKPOINT_FILE))
@@ -210,7 +227,8 @@ final class LogManagerTest extends LogTestBase {
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
newLogManager.startup();
logManager = newLogManager;
log1 = getOrCreateLog(tablePath1, null, tableBucket1);
@@ -255,7 +273,8 @@ final class LogManagerTest extends LogTestBase {
zkClient,
new FlussScheduler(1),
SystemClock.getInstance(),
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
assertThat(new File(dataDir, CLEAN_SHUTDOWN_FILE).exists()).isTrue();
newLogManager.startup();
logManager = newLogManager;
@@ -300,9 +319,84 @@ final class LogManagerTest extends LogTestBase {
assertThat(logManager.getLog(log1.getTableBucket()).isPresent()).isTrue();
}
+ @Test
+ @Tag(ServerTestTags.JBOD_MULTI_DIR_TAG)
+ void testCheckpointRecoveryPointsAreWrittenPerDirectory() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ initTableBuckets(null);
+
+ LogTablet log1 = createLog(tablePath1, tableBucket1, dataDir1);
+ LogTablet log2 = createLog(tablePath2, tableBucket2, dataDir2);
+
+ MemoryLogRecords records = genMemoryLogRecordsByObject(DATA1);
+ log1.appendAsLeader(records);
+ log2.appendAsLeader(records);
+ log1.flush(false);
+ log2.flush(false);
+
+ logManager.checkpointRecoveryOffsets(dataDir1);
+ logManager.checkpointRecoveryOffsets(dataDir2);
+
+ Map<TableBucket, Long> dir1Checkpoints =
+ new OffsetCheckpointFile(
+ new File(dataDir1, LogManager.RECOVERY_POINT_CHECKPOINT_FILE))
+ .read();
+ Map<TableBucket, Long> dir2Checkpoints =
+ new OffsetCheckpointFile(
+ new File(dataDir2, LogManager.RECOVERY_POINT_CHECKPOINT_FILE))
+ .read();
+
+ assertThat(dir1Checkpoints).containsOnlyKeys(tableBucket1);
+ assertThat(dir1Checkpoints.get(tableBucket1)).isEqualTo(log1.getRecoveryPoint());
+ assertThat(dir2Checkpoints).containsOnlyKeys(tableBucket2);
+ assertThat(dir2Checkpoints.get(tableBucket2)).isEqualTo(log2.getRecoveryPoint());
+ }
+
+ @Test
+ @Tag(ServerTestTags.JBOD_MULTI_DIR_TAG)
+ void testPerDirectoryCleanShutdownAndRecovery() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ initTableBuckets(null);
+
+ createLog(tablePath1, tableBucket1, dataDir1)
+ .appendAsLeader(genMemoryLogRecordsByObject(DATA1));
+ createLog(tablePath2, tableBucket2, dataDir2)
+ .appendAsLeader(genMemoryLogRecordsByObject(DATA1));
+
+ logManager.shutdown();
+ logManager = null;
+ localDiskManager.close();
+
+ assertThat(new File(dataDir1, LogManager.CLEAN_SHUTDOWN_FILE)).exists();
+ assertThat(new File(dataDir2, LogManager.CLEAN_SHUTDOWN_FILE)).exists();
+
+ localDiskManager = LocalDiskManager.create(conf);
+ logManager =
+ LogManager.create(
+ conf,
+ zkClient,
+ new FlussScheduler(1),
+ SystemClock.getInstance(),
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
+ logManager.startup();
+
+ assertThat(new File(dataDir1, LogManager.CLEAN_SHUTDOWN_FILE)).doesNotExist();
+ assertThat(new File(dataDir2, LogManager.CLEAN_SHUTDOWN_FILE)).doesNotExist();
+ assertThat(logManager.getLog(tableBucket1)).isPresent();
+ assertThat(logManager.getLog(tableBucket1).get().getDataDir())
+ .isEqualTo(dataDir1.getAbsoluteFile());
+ assertThat(logManager.getLog(tableBucket2)).isPresent();
+ assertThat(logManager.getLog(tableBucket2).get().getDataDir())
+ .isEqualTo(dataDir2.getAbsoluteFile());
+ }
+
private LogTablet getOrCreateLog(
TablePath tablePath, String partitionName, TableBucket tableBucket) throws Exception {
return logManager.getOrCreateLog(
+ tempDir,
PhysicalTablePath.of(
tablePath.getDatabaseName(), tablePath.getTableName(), partitionName),
tableBucket,
@@ -311,6 +405,12 @@ final class LogManagerTest extends LogTestBase {
false);
}
+ private LogTablet createLog(TablePath tablePath, TableBucket tableBucket, File dataDir)
+ throws Exception {
+ return logManager.getOrCreateLog(
+ dataDir, PhysicalTablePath.of(tablePath), tableBucket, LogFormat.ARROW, 1, false);
+ }
+
private void initTableBuckets(@Nullable String partitionName) {
if (partitionName == null) {
tableBucket1 = new TableBucket(DATA1_TABLE_ID, 1);
@@ -326,9 +426,12 @@ final class LogManagerTest extends LogTestBase {
}
@AfterEach
- public void tearDown() {
+ public void tearDown() throws Exception {
if (logManager != null) {
logManager.shutdown();
}
+ if (localDiskManager != null) {
+ localDiskManager.close();
+ }
}
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
index 2ef483205..46685cfe0 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
@@ -88,6 +88,7 @@ final class LogTabletTest extends LogTestBase {
scheduler.startup();
logTablet =
LogTablet.create(
+ tempDir,
PhysicalTablePath.of(DATA1_TABLE_PATH),
logDir,
conf,
@@ -491,6 +492,7 @@ final class LogTabletTest extends LogTestBase {
Scheduler scheduler = new FlussScheduler(1);
scheduler.startup();
return LogTablet.create(
+ tempDir,
PhysicalTablePath.of(DATA1_TABLE_PATH),
logDir,
config,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogIndexCacheTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogIndexCacheTest.java
index 841e58a13..d6788c495 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogIndexCacheTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogIndexCacheTest.java
@@ -49,7 +49,7 @@ class RemoteLogIndexCacheTest extends RemoteLogTestBase {
@BeforeEach
public void setup() throws Exception {
super.setup();
- rlIndexCache = remoteLogManager.getRemoteLogIndexCache();
+ rlIndexCache = remoteLogManager.getRemoteLogIndexCache(localDiskManager.dataDirs().get(0));
}
@ParameterizedTest
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java
index 2ac4e2096..2944f1c66 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server.log.remote;
import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.exception.NotLeaderOrFollowerException;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.remote.RemoteLogFetchInfo;
@@ -32,14 +33,19 @@ import org.apache.fluss.server.log.FetchParams;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.replica.Replica;
import org.apache.fluss.server.replica.ReplicaManager;
+import org.apache.fluss.server.testutils.ServerTestTags;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.io.File;
import java.time.Duration;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -605,6 +611,28 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
.isEqualTo(-1L);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testLookupRemoteIndexThrowsWhenLocalLogMissing(boolean partitionTable) throws Exception {
+ TableBucket tb = makeTableBucket(partitionTable);
+ makeLogTableAsLeader(tb, partitionTable);
+ LogTablet logTablet = replicaManager.getReplicaOrException(tb).getLogTablet();
+ long startTimestamp = manualClock.milliseconds();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ RemoteLogSegment remoteLogSegment =
+ remoteLogManager.relevantRemoteLogSegments(tb, 0L).get(0);
+ logManager.dropLog(tb);
+
+ assertThatThrownBy(() -> remoteLogManager.lookupPositionForOffset(remoteLogSegment, 2L))
+ .isInstanceOf(NotLeaderOrFollowerException.class)
+ .hasMessageContaining("Can't resolve remote log index cache for bucket " + tb);
+ assertThatThrownBy(() -> remoteLogManager.lookupOffsetForTimestamp(tb, startTimestamp))
+ .isInstanceOf(NotLeaderOrFollowerException.class)
+ .hasMessageContaining("Can't resolve remote log index cache for bucket " + tb);
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testAlterTableTieredLogLocalSegments(boolean partitionedTable) throws Exception {
@@ -711,4 +739,49 @@ class RemoteLogManagerTest extends RemoteLogTestBase {
Arguments.of(true, false),
Arguments.of(true, true));
}
+
+ // ---- JBOD multi-directory tests ----
+
+ @Test
+ @Tag(ServerTestTags.JBOD_MULTI_DIR_TAG)
+ void testRemoteIndexCacheFollowsReplicaDirectory() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ makeLogTableAsLeader(0);
+ makeLogTableAsLeader(1);
+
+ TableBucket tableBucket1 = new TableBucket(DATA1_TABLE_ID, 0);
+ TableBucket tableBucket2 = new TableBucket(DATA1_TABLE_ID, 1);
+ Replica replica1 = replicaManager.getReplicaOrException(tableBucket1);
+ Replica replica2 = replicaManager.getReplicaOrException(tableBucket2);
+ assertThat(replica1.getLogTablet().getDataDir()).isEqualTo(dataDir1.getAbsoluteFile());
+ assertThat(replica2.getLogTablet().getDataDir()).isEqualTo(dataDir2.getAbsoluteFile());
+
+ addMultiSegmentsToLogTablet(replica1.getLogTablet(), 5);
+ addMultiSegmentsToLogTablet(replica2.getLogTablet(), 5);
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ RemoteLogSegment segment1 =
+ remoteLogManager.relevantRemoteLogSegments(tableBucket1, 0L).get(0);
+ RemoteLogSegment segment2 =
+ remoteLogManager.relevantRemoteLogSegments(tableBucket2, 0L).get(0);
+ remoteLogManager.lookupPositionForOffset(segment1, 2L);
+ remoteLogManager.lookupPositionForOffset(segment2, 2L);
+
+ List<String> dir1Files =
+ Arrays.stream(remoteLogManager.getRemoteLogIndexCache(dataDir1).cacheDir().list())
+ .collect(Collectors.toList());
+ List<String> dir2Files =
+ Arrays.stream(remoteLogManager.getRemoteLogIndexCache(dataDir2).cacheDir().list())
+ .collect(Collectors.toList());
+
+ assertThat(dir1Files)
+ .anyMatch(name -> name.contains(segment1.remoteLogSegmentId().toString()));
+ assertThat(dir1Files)
+ .noneMatch(name -> name.contains(segment2.remoteLogSegmentId().toString()));
+ assertThat(dir2Files)
+ .anyMatch(name -> name.contains(segment2.remoteLogSegmentId().toString()));
+ assertThat(dir2Files)
+ .noneMatch(name -> name.contains(segment1.remoteLogSegmentId().toString()));
+ }
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
index fa502f928..14e24c18c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
@@ -68,12 +68,14 @@ import org.apache.fluss.server.kv.rocksdb.RocksDBKv;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
import org.apache.fluss.server.log.FetchParams;
import org.apache.fluss.server.log.ListOffsetsParam;
+import org.apache.fluss.server.log.checkpoint.OffsetCheckpointFile;
import org.apache.fluss.server.metadata.BucketMetadata;
import org.apache.fluss.server.metadata.ClusterMetadata;
import org.apache.fluss.server.metadata.PartitionMetadata;
import org.apache.fluss.server.metadata.ServerInfo;
import org.apache.fluss.server.metadata.TableMetadata;
import org.apache.fluss.server.testutils.KvTestUtils;
+import org.apache.fluss.server.testutils.ServerTestTags;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.apache.fluss.server.zk.data.TableRegistration;
@@ -84,12 +86,14 @@ import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.CloseableIterator;
import org.apache.fluss.utils.types.Tuple2;
+import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
+import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -132,6 +136,7 @@ import static org.apache.fluss.record.TestData.EXPECTED_LOG_RESULTS_FOR_DATA_1_W
import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH;
import static org.apache.fluss.server.metadata.PartitionMetadata.DELETED_PARTITION_ID;
import static org.apache.fluss.server.metadata.TableMetadata.DELETED_TABLE_ID;
+import static org.apache.fluss.server.replica.ReplicaManager.HIGH_WATERMARK_CHECKPOINT_FILE_NAME;
import static org.apache.fluss.server.testutils.PartitionMetadataAssert.assertPartitionMetadata;
import static org.apache.fluss.server.testutils.TableMetadataAssert.assertTableMetadata;
import static org.apache.fluss.server.zk.data.LeaderAndIsr.INITIAL_BUCKET_EPOCH;
@@ -2413,4 +2418,70 @@ class ReplicaManagerTest extends ReplicaTestBase {
assertThat(scannerManager.activeScannerCount()).isEqualTo(0);
}
+
+ // ---- JBOD multi-directory tests ----
+
+ @Test
+ @Tag(ServerTestTags.JBOD_MULTI_DIR_TAG)
+ void testNewBucketsDistributedAcrossDataDirs() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ makeLogTableAsLeader(0);
+ makeLogTableAsLeader(1);
+
+ Replica logReplica1 =
+ replicaManager.getReplicaOrException(new TableBucket(DATA1_TABLE_ID, 0));
+ Replica logReplica2 =
+ replicaManager.getReplicaOrException(new TableBucket(DATA1_TABLE_ID, 1));
+ assertThat(logReplica1.getLogTablet().getDataDir()).isEqualTo(dataDir1.getAbsoluteFile());
+ assertThat(logReplica2.getLogTablet().getDataDir()).isEqualTo(dataDir2.getAbsoluteFile());
+ }
+
+ @Test
+ @Tag(ServerTestTags.JBOD_MULTI_DIR_TAG)
+ void testLogAndKvCoLocatedForPrimaryKeyTable() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ makeKvTableAsLeader(DATA1_TABLE_ID_PK, DATA1_TABLE_PATH_PK, 0);
+ makeKvTableAsLeader(DATA1_TABLE_ID_PK, DATA1_TABLE_PATH_PK, 1);
+
+ Replica kvReplica1 =
+ replicaManager.getReplicaOrException(new TableBucket(DATA1_TABLE_ID_PK, 0));
+ Replica kvReplica2 =
+ replicaManager.getReplicaOrException(new TableBucket(DATA1_TABLE_ID_PK, 1));
+ assertThat(kvReplica1.getLogTablet().getLogDir().toPath()).startsWith(dataDir1.toPath());
+ assertThat(kvReplica1.getKvTablet().getKvTabletDir().toPath())
+ .startsWith(dataDir1.toPath());
+ assertThat(kvReplica2.getLogTablet().getLogDir().toPath()).startsWith(dataDir2.toPath());
+ assertThat(kvReplica2.getKvTablet().getKvTabletDir().toPath())
+ .startsWith(dataDir2.toPath());
+ }
+
+ @Test
+ @Tag(ServerTestTags.JBOD_MULTI_DIR_TAG)
+ void testHighWatermarkCheckpointIsWrittenPerDirectory() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ makeLogTableAsLeader(0);
+ makeLogTableAsLeader(1);
+
+ Replica replica1 = replicaManager.getReplicaOrException(new TableBucket(DATA1_TABLE_ID, 0));
+ Replica replica2 = replicaManager.getReplicaOrException(new TableBucket(DATA1_TABLE_ID, 1));
+ replica1.appendRecordsToLeader(genMemoryLogRecordsByObject(DATA1), 1);
+ replica2.appendRecordsToLeader(genMemoryLogRecordsByObject(DATA1), 1);
+
+ replicaManager.checkpointHighWatermarks();
+
+ Map<TableBucket, Long> checkpoint1 =
+ new OffsetCheckpointFile(new File(dataDir1, HIGH_WATERMARK_CHECKPOINT_FILE_NAME))
+ .read();
+ Map<TableBucket, Long> checkpoint2 =
+ new OffsetCheckpointFile(new File(dataDir2, HIGH_WATERMARK_CHECKPOINT_FILE_NAME))
+ .read();
+
+ assertThat(checkpoint1).containsOnlyKeys(new TableBucket(DATA1_TABLE_ID, 0));
+ assertThat(checkpoint1.get(new TableBucket(DATA1_TABLE_ID, 0))).isEqualTo(10L);
+ assertThat(checkpoint2).containsOnlyKeys(new TableBucket(DATA1_TABLE_ID, 1));
+ assertThat(checkpoint2.get(new TableBucket(DATA1_TABLE_ID, 1))).isEqualTo(10L);
+ }
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
index a9041acbd..bad818a9e 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
@@ -54,6 +54,8 @@ import org.apache.fluss.server.metadata.ServerInfo;
import org.apache.fluss.server.metadata.TabletServerMetadataCache;
import org.apache.fluss.server.metrics.group.BucketMetricGroup;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
+import org.apache.fluss.server.storage.LocalDiskManager;
+import org.apache.fluss.server.testutils.ServerTestTags;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
@@ -70,6 +72,7 @@ import org.apache.fluss.utils.function.ThrowingRunnable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
@@ -125,6 +128,7 @@ import static org.apache.fluss.utils.FlussPaths.remoteLogTabletDir;
* function managed by {@link ReplicaManager}.
*/
public class ReplicaTestBase {
+
@RegisterExtension
public static final AllCallbackWrapper<ZooKeeperExtension> ZOO_KEEPER_EXTENSION_WRAPPER =
new AllCallbackWrapper<>(new ZooKeeperExtension());
@@ -138,6 +142,7 @@ public class ReplicaTestBase {
protected @TempDir File tempDir;
protected ManualClock manualClock;
+ protected LocalDiskManager localDiskManager;
protected LogManager logManager;
protected KvManager kvManager;
protected ReplicaManager replicaManager;
@@ -170,9 +175,18 @@ public class ReplicaTestBase {
}
@BeforeEach
- public void setup() throws Exception {
+ public void setup(TestInfo testInfo) throws Exception {
conf = getServerConf();
- conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath());
+ conf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
+ if (testInfo != null && testInfo.getTags().contains(ServerTestTags.JBOD_MULTI_DIR_TAG)) {
+ conf.set(
+ ConfigOptions.DATA_DIRS,
+ Arrays.asList(
+ new File(tempDir, "data-1").getAbsolutePath(),
+ new File(tempDir, "data-2").getAbsolutePath()));
+ } else {
+ conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath());
+ }
conf.setString(ConfigOptions.COORDINATOR_HOST, "localhost");
conf.set(ConfigOptions.REMOTE_DATA_DIR, tempDir.getAbsolutePath() + "/remote_data_dir");
conf.set(ConfigOptions.SERVER_IO_POOL_SIZE, 2);
@@ -190,18 +204,24 @@ public class ReplicaTestBase {
ioExecutor = Executors.newSingleThreadExecutor();
manualClock = new ManualClock(System.currentTimeMillis());
+ localDiskManager = LocalDiskManager.create(conf);
logManager =
LogManager.create(
conf,
zkClient,
scheduler,
manualClock,
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
logManager.startup();
kvManager =
KvManager.create(
- conf, zkClient, logManager, TestingMetricGroups.TABLET_SERVER_METRICS);
+ conf,
+ zkClient,
+ logManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
kvManager.startup();
serverMetadataCache =
@@ -227,6 +247,10 @@ public class ReplicaTestBase {
registerTableInZkClient();
}
+ // Kept for subclasses that still define @BeforeEach setup() and call super.setup().
+ // The actual initialization already happens in setup(TestInfo).
+ protected void setup() throws Exception {}
+
private void initMetadataCache(TabletServerMetadataCache metadataCache) {
metadataCache.updateClusterMetadata(
new ClusterMetadata(
@@ -330,7 +354,8 @@ public class ReplicaTestBase {
remoteLogManager,
scannerManager,
manualClock,
- ioExecutor);
+ ioExecutor,
+ localDiskManager);
}
@AfterEach
@@ -369,6 +394,10 @@ public class ReplicaTestBase {
scheduler.shutdown();
}
+ if (localDiskManager != null) {
+ localDiskManager.close();
+ }
+
if (ioExecutor != null) {
ioExecutor.shutdown();
}
@@ -491,6 +520,7 @@ public class ReplicaTestBase {
.getServerMetricGroup()
.addTableBucketMetricGroup(physicalTablePath, tableBucket, isPkTable);
return new Replica(
+ localDiskManager.selectDataDirForNewBucket(isPkTable),
physicalTablePath,
tableBucket,
logManager,
@@ -501,7 +531,7 @@ public class ReplicaTestBase {
new OffsetCheckpointFile.LazyOffsetCheckpoints(
new OffsetCheckpointFile(
new File(
- conf.getString(ConfigOptions.DATA_DIR),
+ localDiskManager.dataDirs().get(0),
HIGH_WATERMARK_CHECKPOINT_FILE_NAME))),
replicaManager.getDelayedWriteManager(),
replicaManager.getDelayedFetchLogManager(),
@@ -524,6 +554,8 @@ public class ReplicaTestBase {
conf,
zkClient,
testCoordinatorGateway,
+ localDiskManager,
+ logManager,
remoteLogStorage,
remoteLogTaskScheduler,
manualClock);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
index b68cbe6b7..5dc17acf1 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
@@ -41,6 +41,7 @@ import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
import org.apache.fluss.server.replica.Replica;
import org.apache.fluss.server.replica.ReplicaManager;
+import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
@@ -105,6 +106,8 @@ public class ReplicaFetcherThreadTest {
private ReplicaManager followerRM;
private ReplicaFetcherThread followerFetcher;
private ExecutorService ioExecutor;
+ private LocalDiskManager leaderLocalDiskManager;
+ private LocalDiskManager followerLocalDiskManager;
@BeforeAll
static void baseBeforeAll() {
@@ -120,8 +123,10 @@ public class ReplicaFetcherThreadTest {
manualClock = new ManualClock(System.currentTimeMillis());
Configuration conf = new Configuration();
tb = new TableBucket(DATA1_TABLE_ID, 0);
- leaderRM = createReplicaManager(leaderServerId);
- followerRM = createReplicaManager(followerServerId);
+ leaderLocalDiskManager = createLocalDiskManager(leaderServerId);
+ leaderRM = createReplicaManager(leaderServerId, leaderLocalDiskManager);
+ followerLocalDiskManager = createLocalDiskManager(followerServerId);
+ followerRM = createReplicaManager(followerServerId, followerLocalDiskManager);
// with local test leader end point.
leader =
new ServerNode(
@@ -144,6 +149,12 @@ public class ReplicaFetcherThreadTest {
@AfterEach
public void tearDown() throws Exception {
+ if (leaderLocalDiskManager != null) {
+ leaderLocalDiskManager.close();
+ }
+ if (followerLocalDiskManager != null) {
+ followerLocalDiskManager.close();
+ }
if (ioExecutor != null) {
ioExecutor.shutdownNow();
}
@@ -499,8 +510,17 @@ public class ReplicaFetcherThreadTest {
result -> {});
}
- private ReplicaManager createReplicaManager(int serverId) throws Exception {
+ private LocalDiskManager createLocalDiskManager(int serverId) throws Exception {
Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ID, serverId);
+ conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath() + "/server-" + serverId);
+ return LocalDiskManager.create(conf);
+ }
+
+ private ReplicaManager createReplicaManager(int serverId, LocalDiskManager localDiskManager)
+ throws Exception {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ID, serverId);
conf.setString(ConfigOptions.DATA_DIR, tempDir.getAbsolutePath() + "/server-" + serverId);
conf.set(ConfigOptions.WRITER_ID_EXPIRATION_TIME, Duration.ofHours(12));
Scheduler scheduler = new FlussScheduler(2);
@@ -512,7 +532,8 @@ public class ReplicaFetcherThreadTest {
zkClient,
scheduler,
manualClock,
- TestingMetricGroups.TABLET_SERVER_METRICS);
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager);
logManager.startup();
ReplicaManager replicaManager =
new TestingReplicaManager(
@@ -530,7 +551,8 @@ public class ReplicaFetcherThreadTest {
RpcClient.create(conf, TestingClientMetricGroup.newInstance()),
TestingMetricGroups.TABLET_SERVER_METRICS,
manualClock,
- ioExecutor);
+ ioExecutor,
+ localDiskManager);
replicaManager.startup();
return replicaManager;
}
@@ -551,7 +573,8 @@ public class ReplicaFetcherThreadTest {
RpcClient rpcClient,
TabletServerMetricGroup serverMetricGroup,
Clock clock,
- ExecutorService ioExecutor)
+ ExecutorService ioExecutor,
+ LocalDiskManager localDiskManager)
throws IOException {
super(
conf,
@@ -569,7 +592,8 @@ public class ReplicaFetcherThreadTest {
USER_METRICS,
new ScannerManager(conf, scheduler),
clock,
- ioExecutor);
+ ioExecutor,
+ localDiskManager);
}
@Override
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java
new file mode 100644
index 000000000..78c5bcfd5
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java
@@ -0,0 +1,400 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.storage;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.IllegalConfigurationException;
+import org.apache.fluss.exception.LogStorageException;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.util.Arrays;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link LocalDiskManager}. */
+class LocalDiskManagerTest {
+
+ private static final int TABLET_SERVER_ID = 1;
+
+ @TempDir private File tempDir;
+
+ @Test
+ void testCreateDiskPropertiesAndReuseDiskId() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ Configuration conf = createConf(dataDir1, dataDir2);
+
+ String diskId1;
+ String diskId2;
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ diskId1 = diskId(dataDir1);
+ diskId2 = diskId(dataDir2);
+ assertThat(new File(dataDir1, LocalDiskManager.DISK_PROPERTIES_FILE_NAME)).exists();
+ assertThat(new File(dataDir2, LocalDiskManager.DISK_PROPERTIES_FILE_NAME)).exists();
+ assertThat(new File(dataDir1, LocalDiskManager.LOCK_FILE_NAME)).exists();
+ assertThat(new File(dataDir2, LocalDiskManager.LOCK_FILE_NAME)).exists();
+ assertThat(localDiskManager.diskId(dataDir1)).isEqualTo(diskId1);
+ assertThat(localDiskManager.diskId(dataDir2)).isEqualTo(diskId2);
+ }
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ assertThat(localDiskManager.diskId(dataDir1)).isEqualTo(diskId1);
+ assertThat(localDiskManager.diskId(dataDir2)).isEqualTo(diskId2);
+ }
+ }
+
+ @Test
+ void testFallbackToDataDirWhenDataDirsIsNotConfigured() throws Exception {
+ File expectedDataDir = new File(tempDir, "nested/../data-single");
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
+ conf.setString(ConfigOptions.DATA_DIR, expectedDataDir.getPath());
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ assertThat(localDiskManager.dataDirs())
+ .containsExactly(
+ expectedDataDir.getAbsoluteFile().toPath().normalize().toFile());
+ }
+ }
+
+ @Test
+ void testDataDirsTakePrecedenceOverDataDir() throws Exception {
+ File fallbackDataDir = new File(tempDir, "fallback-single");
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
+ conf.setString(ConfigOptions.DATA_DIR, fallbackDataDir.getAbsolutePath());
+ conf.set(
+ ConfigOptions.DATA_DIRS,
+ Arrays.asList(dataDir1.getAbsolutePath(), dataDir2.getAbsolutePath()));
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ assertThat(localDiskManager.dataDirs())
+ .containsExactly(dataDir1.getAbsoluteFile(), dataDir2.getAbsoluteFile());
+ assertThat(fallbackDataDir).doesNotExist();
+ }
+ }
+
+ @Test
+ void testNormalizeConfiguredDataDirPaths() throws Exception {
+ File normalizedDataDir = new File(tempDir, "data-1");
+ Configuration normalizedConf =
+ createConfFromPaths(new File(tempDir, "nested/../data-1").getPath());
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(normalizedConf)) {
+ assertThat(localDiskManager.dataDirs())
+ .containsExactly(normalizedDataDir.getAbsoluteFile());
+ }
+ }
+
+ @Test
+ void testRejectDuplicateAndOverlappingDirs() {
+ // exact duplicate
+ Configuration duplicateConf = new Configuration();
+ duplicateConf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
+ String sameDir = new File(tempDir, "same").getAbsolutePath();
+ duplicateConf.set(ConfigOptions.DATA_DIRS, Arrays.asList(sameDir, sameDir));
+
+ assertThatThrownBy(() -> LocalDiskManager.create(duplicateConf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("Duplicate local data directory");
+
+ // normalized duplicate
+ File normalizedDataDir = new File(tempDir, "data-1");
+ Configuration normalizedDupConf =
+ createConfFromPaths(
+ normalizedDataDir.getAbsolutePath(),
+ new File(tempDir, "nested/../data-1").getPath());
+ assertThatThrownBy(() -> LocalDiskManager.create(normalizedDupConf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("Duplicate local data directory");
+
+ // parent-child overlap
+ File parent = new File(tempDir, "parent");
+ Configuration overlapConf = new Configuration();
+ overlapConf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
+ overlapConf.set(
+ ConfigOptions.DATA_DIRS,
+ Arrays.asList(
+ parent.getAbsolutePath(), new File(parent, "child").getAbsolutePath()));
+
+ assertThatThrownBy(() -> LocalDiskManager.create(overlapConf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("must not overlap");
+
+ // normalized overlap
+ Configuration normalizedOverlapConf =
+ createConfFromPaths(
+ new File(tempDir, "root").getAbsolutePath(),
+ new File(tempDir, "nested/../root/child").getPath());
+ assertThatThrownBy(() -> LocalDiskManager.create(normalizedOverlapConf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("must not overlap");
+ }
+
+ @Test
+ void testDirectoryLocksPreventDoubleOwnership() throws Exception {
+ File dataDir = new File(tempDir, "locked");
+ Configuration conf = createConf(dataDir);
+
+ try (LocalDiskManager ignored = LocalDiskManager.create(conf)) {
+ assertThatThrownBy(() -> LocalDiskManager.create(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("Failed to acquire lock on file .lock")
+ .hasMessageContaining("another process or thread is using this directory");
+ }
+ }
+
+ @Test
+ void testRejectDuplicateDiskIdsAcrossDataDirs() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ assertThat(dataDir1.mkdirs()).isTrue();
+ assertThat(dataDir2.mkdirs()).isTrue();
+ writeDiskProperties(dataDir1, "shared-disk-id", TABLET_SERVER_ID);
+ writeDiskProperties(dataDir2, "shared-disk-id", TABLET_SERVER_ID);
+
+ Configuration conf = createConf(dataDir1, dataDir2);
+
+ assertThatThrownBy(() -> LocalDiskManager.create(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("duplicate disk.id");
+ assertThat(new File(dataDir1, LocalDiskManager.LOCK_FILE_NAME)).doesNotExist();
+ assertThat(new File(dataDir2, LocalDiskManager.LOCK_FILE_NAME)).doesNotExist();
+ }
+
+ @Test
+ void testKeepHealthyDirWhenAnotherDirHasBrokenDiskProperties() throws Exception {
+ File brokenDataDir = new File(tempDir, "broken-data");
+ File healthyDataDir = new File(tempDir, "healthy-data");
+ assertThat(brokenDataDir.mkdirs()).isTrue();
+ assertThat(healthyDataDir.mkdirs()).isTrue();
+
+ Properties properties = new Properties();
+ properties.setProperty("version", "999");
+ properties.setProperty("disk.id", "broken-disk-id");
+ properties.setProperty("server.id", String.valueOf(TABLET_SERVER_ID));
+ try (FileOutputStream outputStream =
+ new FileOutputStream(
+ new File(brokenDataDir, LocalDiskManager.DISK_PROPERTIES_FILE_NAME))) {
+ properties.store(outputStream, null);
+ }
+
+ Configuration conf = createConf(brokenDataDir, healthyDataDir);
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ assertThat(localDiskManager.dataDirs())
+ .containsExactly(healthyDataDir.getAbsoluteFile());
+ assertThat(localDiskManager.diskId(healthyDataDir)).isNotBlank();
+ assertThatThrownBy(() -> localDiskManager.diskId(brokenDataDir))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+
+ @Test
+ void testKeepHealthyDirWhenAnotherDirCannotBeValidated() throws Exception {
+ File badPath = new File(tempDir, "not-a-dir");
+ assertThat(badPath.createNewFile()).isTrue();
+ File healthyDataDir = new File(tempDir, "healthy-data");
+
+ Configuration conf = createConf(badPath, healthyDataDir);
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ assertThat(localDiskManager.dataDirs())
+ .containsExactly(healthyDataDir.getAbsoluteFile());
+ assertThat(new File(healthyDataDir, LocalDiskManager.DISK_PROPERTIES_FILE_NAME))
+ .exists();
+ assertThat(new File(healthyDataDir, LocalDiskManager.LOCK_FILE_NAME)).exists();
+ }
+ }
+
+ @Test
+ void testSelectDataDirUsesConfiguredOrderAsTieBreaker() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ Configuration conf = createConf(dataDir2, dataDir1);
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ assertThat(localDiskManager.selectDataDirForNewBucket(false))
+ .isEqualTo(dataDir2.getAbsoluteFile());
+ assertThat(localDiskManager.selectDataDirForNewBucket(true))
+ .isEqualTo(dataDir2.getAbsoluteFile());
+ }
+ }
+
+ @Test
+ void testSelectDataDirUsesSeparatedLogAndKvCounters() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ Configuration conf = createConf(dataDir1, dataDir2);
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ File logDataDir1 = localDiskManager.selectDataDirForNewBucket(false);
+ assertThat(logDataDir1).isEqualTo(dataDir1.getAbsoluteFile());
+ localDiskManager.recordReplicaLoad(logDataDir1, false);
+
+ File logDataDir2 = localDiskManager.selectDataDirForNewBucket(false);
+ assertThat(logDataDir2).isEqualTo(dataDir2.getAbsoluteFile());
+ localDiskManager.recordReplicaLoad(logDataDir2, false);
+
+ File kvDataDir1 = localDiskManager.selectDataDirForNewBucket(true);
+ assertThat(kvDataDir1).isEqualTo(dataDir1.getAbsoluteFile());
+ localDiskManager.recordReplicaLoad(kvDataDir1, true);
+
+ File kvDataDir2 = localDiskManager.selectDataDirForNewBucket(true);
+ assertThat(kvDataDir2).isEqualTo(dataDir2.getAbsoluteFile());
+
+ assertThat(localDiskManager.logBucketCount(dataDir1)).isEqualTo(2);
+ assertThat(localDiskManager.logBucketCount(dataDir2)).isEqualTo(1);
+ assertThat(localDiskManager.kvBucketCount(dataDir1)).isEqualTo(1);
+ assertThat(localDiskManager.kvBucketCount(dataDir2)).isEqualTo(0);
+ }
+ }
+
+ @Test
+ void testRejectMismatchedServerIdInDiskProperties() throws Exception {
+ File dataDir = new File(tempDir, "data-1");
+ assertThat(dataDir.mkdirs()).isTrue();
+ writeDiskProperties(dataDir, "disk-1", TABLET_SERVER_ID + 1);
+
+ assertThatThrownBy(() -> LocalDiskManager.create(createConf(dataDir)))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining("does not match server.id");
+ }
+
+ @Test
+ void testResolveDataDir() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ Configuration conf = createConf(dataDir1, dataDir2);
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ File childOfDir1 = new File(dataDir1, "db/table/bucket-0");
+ assertThat(localDiskManager.resolveDataDir(childOfDir1))
+ .isEqualTo(dataDir1.getAbsoluteFile());
+
+ File childOfDir2 = new File(dataDir2, "db/table/bucket-1");
+ assertThat(localDiskManager.resolveDataDir(childOfDir2))
+ .isEqualTo(dataDir2.getAbsoluteFile());
+
+ File unknownPath = new File(tempDir, "unknown/path");
+ assertThatThrownBy(() -> localDiskManager.resolveDataDir(unknownPath))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("does not belong to any configured data directory");
+ }
+ }
+
+ @Test
+ void testRecordReplicaDeleteDecrementsCounters() throws Exception {
+ File dataDir1 = new File(tempDir, "data-1");
+ File dataDir2 = new File(tempDir, "data-2");
+ Configuration conf = createConf(dataDir1, dataDir2);
+
+ try (LocalDiskManager localDiskManager = LocalDiskManager.create(conf)) {
+ // load some replicas
+ localDiskManager.recordReplicaLoad(dataDir1.getAbsoluteFile(), false);
+ localDiskManager.recordReplicaLoad(dataDir1.getAbsoluteFile(), true);
+ localDiskManager.recordReplicaLoad(dataDir2.getAbsoluteFile(), false);
+
+ assertThat(localDiskManager.logBucketCount(dataDir1)).isEqualTo(2);
+ assertThat(localDiskManager.kvBucketCount(dataDir1)).isEqualTo(1);
+ assertThat(localDiskManager.logBucketCount(dataDir2)).isEqualTo(1);
+
+ // delete log-only replica
+ localDiskManager.recordReplicaDelete(dataDir2.getAbsoluteFile(), false);
+ assertThat(localDiskManager.logBucketCount(dataDir2)).isEqualTo(0);
+ assertThat(localDiskManager.kvBucketCount(dataDir2)).isEqualTo(0);
+
+ // delete PK replica
+ localDiskManager.recordReplicaDelete(dataDir1.getAbsoluteFile(), true);
+ assertThat(localDiskManager.logBucketCount(dataDir1)).isEqualTo(1);
+ assertThat(localDiskManager.kvBucketCount(dataDir1)).isEqualTo(0);
+
+ // ensure counts do not go below zero
+ localDiskManager.recordReplicaDelete(dataDir2.getAbsoluteFile(), true);
+ assertThat(localDiskManager.logBucketCount(dataDir2)).isEqualTo(0);
+ assertThat(localDiskManager.kvBucketCount(dataDir2)).isEqualTo(0);
+ }
+ }
+
+ @Test
+ void testAllDirectoriesUnusableFailsStartup() throws Exception {
+ // create two paths that are regular files, not directories
+ File badPath1 = new File(tempDir, "bad-1");
+ File badPath2 = new File(tempDir, "bad-2");
+ assertThat(badPath1.createNewFile()).isTrue();
+ assertThat(badPath2.createNewFile()).isTrue();
+
+ Configuration conf = createConf(badPath1, badPath2);
+
+ assertThatThrownBy(() -> LocalDiskManager.create(conf))
+ .isInstanceOf(LogStorageException.class)
+ .hasMessageContaining("None of the specified data dirs");
+ }
+
+ private Configuration createConf(File... dataDirs) {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
+ conf.set(
+ ConfigOptions.DATA_DIRS,
+ Arrays.stream(dataDirs).map(File::getAbsolutePath).collect(Collectors.toList()));
+ return conf;
+ }
+
+ private Configuration createConfFromPaths(String... dataDirs) {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.TABLET_SERVER_ID, TABLET_SERVER_ID);
+ conf.set(ConfigOptions.DATA_DIRS, Arrays.asList(dataDirs));
+ return conf;
+ }
+
+ private String diskId(File dataDir) throws Exception {
+ Properties properties = new Properties();
+ try (FileInputStream inputStream =
+ new FileInputStream(
+ new File(dataDir, LocalDiskManager.DISK_PROPERTIES_FILE_NAME))) {
+ properties.load(inputStream);
+ }
+ assertThat(properties.getProperty("version")).isEqualTo("1");
+ assertThat(properties.getProperty("server.id")).isEqualTo(String.valueOf(TABLET_SERVER_ID));
+ return properties.getProperty("disk.id");
+ }
+
+ private void writeDiskProperties(File dataDir, String diskId, int serverId) throws Exception {
+ Properties properties = new Properties();
+ properties.setProperty("version", "1");
+ properties.setProperty("disk.id", diskId);
+ properties.setProperty("server.id", String.valueOf(serverId));
+ try (FileOutputStream outputStream =
+ new FileOutputStream(
+ new File(dataDir, LocalDiskManager.DISK_PROPERTIES_FILE_NAME))) {
+ properties.store(outputStream, null);
+ }
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/ServerTestTags.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/ServerTestTags.java
new file mode 100644
index 000000000..167229d5b
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/ServerTestTags.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.testutils;
+
+/** Common JUnit tags for server tests. */
+public final class ServerTestTags {
+
+ public static final String JBOD_MULTI_DIR_TAG = "jbod-multidir";
+
+ private ServerTestTags() {}
+}