[server] Implement JBOD Phase 1 Local Multi-Directory Support by hanliu0830 · Pull Request #3030 · apache/fluss

hanliu0830 · 2026-04-08T12:54:00Z

(The sections below can be removed for hotfixes or typos)
-->

Purpose

Linked issue: close #145

Brief change log

Tests

API and Format

Documentation

Copilot

Pull request overview

Implements Phase 1 of JBOD (multi local disk) support in the Fluss TabletServer by introducing a LocalDiskManager that validates/locks multiple local directories, persists per-disk identity, and enables per-directory checkpointing + placement decisions across Log/KV/Replica/RemoteLog components.

Changes:

Add data.dirs config option and a new LocalDiskManager that validates/locks directories and manages per-disk metadata + simple load counters.
Plumb multi-directory awareness through TabletServer, LogManager, KvManager, ReplicaManager, Replica, and RemoteLogManager (per-dir checkpoints, placement, cache).
Extend/adjust unit tests to cover multi-directory placement and per-directory checkpoint/cache behavior.

Reviewed changes

Copilot reviewed 18 out of 18 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
fluss-server/src/test/java/org/apache/fluss/server/storage/LocalDiskManagerTest.java	New unit tests for directory validation, locking, disk.properties, and placement counters.
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java	Test base now initializes `LocalDiskManager` and supports tagged JBOD multi-dir tests.
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java	Adds JBOD multi-dir tests for distribution and per-dir high-watermark checkpointing.
fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java	Updates replica-manager construction in tests to pass `LocalDiskManager`.
fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogManagerTest.java	Adds JBOD multi-dir test verifying index cache is tied to replica directory.
fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogIndexCacheTest.java	Uses per-dir `RemoteLogIndexCache` accessor.
fluss-server/src/test/java/org/apache/fluss/server/log/LogManagerTest.java	Adds JBOD tests for per-dir recovery checkpoints and clean shutdown markers.
fluss-server/src/test/java/org/apache/fluss/server/log/DroppedTableRecoveryTest.java	Updates tests to pass explicit data dir and close/recreate `LocalDiskManager`.
fluss-server/src/test/java/org/apache/fluss/server/kv/KvManagerTest.java	Updates KV tests to pass `LocalDiskManager` and explicit data dir for tablet creation.
fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java	Refactors to operate on a list of data dirs; lists tablets per dir; adds per-dir executor helper.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java	Creates/closes `LocalDiskManager` and passes it into log/kv/replica managers.
fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java	New component that owns multi-dir validation, locking, disk identity, and placement counters.
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Per-dir high-watermark checkpoints; placement uses `LocalDiskManager`; remote log manager ctor updated.
fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java	Replica now carries `dataDir` to ensure Log/KV tablets and paths are created in the chosen directory.
fluss-server/src/main/java/org/apache/fluss/server/log/remote/RemoteLogManager.java	Maintains per-dir remote index caches; resolves cache based on bucket/log location.
fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java	Multi-dir recovery/shutdown, per-dir recovery-point checkpoints, and explicit data-dir log creation.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java	Multi-dir tablet creation APIs; plumbs selected data dir into KV tablet creation.
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds new `data.dirs` configuration option.

Comments suppressed due to low confidence (1)

fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java:134

With multiple data.dirs, log recovery/shutdown now creates per-directory thread pools. Because recoveryThreadsPerDataDir is sourced from netty.server.num-worker-threads (default 8), the total thread count scales as numDataDirs * netty.server.num-worker-threads (plus extra per-dir executors), which can be much higher than before. Consider introducing a dedicated recovery thread config (total vs per-dir) or otherwise capping threads to avoid startup/shutdown thread explosion on hosts with many disks.

        return new LogManager(
                localDiskManager,
                conf,
                zkClient,
                conf.getInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS),
                scheduler,

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-04-09T03:15:51Z

    public int lookupPositionForOffset(RemoteLogSegment remoteLogSegment, long offset) {
-        return remoteLogIndexCache.lookupPosition(remoteLogSegment, offset);
+        RemoteLogIndexCache remoteLogIndexCache =
+                remoteLogIndexCacheForBucket(remoteLogSegment.tableBucket());
+        return remoteLogIndexCache == null
+                ? -1


lookupPositionForOffset returns -1 when remoteLogIndexCacheForBucket(...) cannot resolve a cache. That firstStartPos is propagated into RemoteLogFetchInfo and the client uses it as the starting position within the first remote segment; -1 effectively causes the client to read from the beginning of the segment (wrong offset / duplicates). Consider failing fast (throw) or providing a deterministic fallback cache so a valid position is always computed for buckets that have remote segments.

Copilot · 2026-04-09T03:15:51Z

+        for (File dataDir : localDiskManager.dataDirs()) {
+            remoteLogIndexCachesByDir.put(
+                    dataDir,
+                    new RemoteLogIndexCache(
+                            (int)
+                                    conf.get(ConfigOptions.REMOTE_LOG_INDEX_FILE_CACHE_SIZE)
+                                            .getBytes(),


REMOTE_LOG_INDEX_FILE_CACHE_SIZE is applied per configured dataDir by creating one RemoteLogIndexCache per directory, which multiplies the effective on-disk cache budget by #data.dirs (and changes the meaning of the config option, whose description implies a total budget). Consider splitting the configured cache size across directories or updating the configuration semantics/documentation accordingly.

Suggested change

for (File dataDir : localDiskManager.dataDirs()) {

remoteLogIndexCachesByDir.put(

dataDir,

new RemoteLogIndexCache(

(int)

conf.get(ConfigOptions.REMOTE_LOG_INDEX_FILE_CACHE_SIZE)

.getBytes(),

List<File> dataDirs = localDiskManager.dataDirs();

long totalRemoteLogIndexFileCacheSizeBytes =

conf.get(ConfigOptions.REMOTE_LOG_INDEX_FILE_CACHE_SIZE).getBytes();

int perDataDirRemoteLogIndexFileCacheSizeBytes =

dataDirs.isEmpty()

? (int) totalRemoteLogIndexFileCacheSizeBytes

: (int) (totalRemoteLogIndexFileCacheSizeBytes / dataDirs.size());

for (File dataDir : dataDirs) {

remoteLogIndexCachesByDir.put(

dataDir,

new RemoteLogIndexCache(

perDataDirRemoteLogIndexFileCacheSizeBytes,

Copilot · 2026-04-09T03:15:52Z

+                    String serverId = properties.getProperty(SERVER_ID_KEY);
+                    return new DiskProperties(parsedVersion, diskId, serverId);
+                } catch (NoSuchFileException e) {
+                    LOG.warn("No disk.properties file under dir {}", file.getAbsolutePath());


Missing disk.properties appears to be an expected state on first startup for a fresh directory, but this is logged at WARN. This can create noisy logs during normal bootstraps; consider lowering this to INFO/DEBUG (or only warn when the directory is non-empty / previously initialized).

Suggested change

LOG.warn("No disk.properties file under dir {}", file.getAbsolutePath());

LOG.info("No disk.properties file under dir {}", file.getAbsolutePath());

Copilot · 2026-04-09T03:15:52Z

 /** Test for {@link LogManager}. */
 final class LogManagerTest extends LogTestBase {
+
+    private static final String JBOD_MULTI_DIR_TAG = "jbod-multidir";


JBOD_MULTI_DIR_TAG duplicates the same tag constant already defined in ReplicaTestBase (ReplicaTestBase.JBOD_MULTI_DIR_TAG). Reusing the shared constant would avoid accidental drift if the tag string changes.

Suggested change

private static final String JBOD_MULTI_DIR_TAG = "jbod-multidir";

private static final String JBOD_MULTI_DIR_TAG =

org.apache.fluss.server.replica.ReplicaTestBase.JBOD_MULTI_DIR_TAG;

gyang94 · 2026-04-20T02:25:58Z

+                }
            }
+        } catch (LogStorageException e) {
+            throw e;


This exception has been thrown inside the loop, the outer try-catch could be removed.

This exception has been thrown inside the loop, the outer try-catch could be removed.

Addressed in 6999b9e. I removed the redundant outer try-catch since the exception is already handled inside the loop.

gyang94 · 2026-04-20T02:37:17Z

-            return remoteLogIndexCache.lookupOffsetForTimestamp(segment, timestamp);
+            RemoteLogIndexCache remoteLogIndexCache = remoteLogIndexCacheForBucket(tableBucket);
+            return remoteLogIndexCache == null
+                    ? -1L


When remoteLogIndexCache is null, will it be better to throw an exception instead of return -1?
The same returning value -1 will mix up the cases between 1. failing to find the offset index. and 2. indexCache is null.

When remoteLogIndexCache is null, will it be better to throw an exception instead of return -1? The same returning value -1 will mix up the cases between 1. failing to find the offset index. and 2. indexCache is null.

Addressed in 6999b9e. I changed this branch to throw NotLeaderOrFollowerException instead of returning -1, so we no longer mix up “index lookup miss” with “failed to resolve the index cache”.

gyang94 · 2026-04-20T02:45:22Z

+    public synchronized File resolveDataDir(File path) {
+        Path pathToResolve = path.toPath();
+        for (File dataDir : dataDirs) {
+            if (pathToResolve.startsWith(dataDir.toPath())) {


should ensure that the pathToResolve should be an absolute path, since we check with startsWith function. pathToResolve.toAbsolutePath().normalize()

should ensure that the pathToResolve should be an absolute path, since we check with startsWith function. pathToResolve.toAbsolutePath().normalize()

Thanks. Addressed in 6999b9e.

gyang94 · 2026-04-20T03:08:56Z

+            }
+
+            try {
+                return Integer.parseInt(serverId.trim());


many trim() calls in one function. In the constructor the serverId already trimmed. Or just trim() once in one function.

many trim() calls in one function. In the constructor the serverId already trimmed. Or just trim() once in one function.

Thanks. Addressed in 6999b9e.

gyang94 · 2026-04-20T03:09:52Z

+        }
+
+        private int serverIdAsInt(File dataDir) throws IOException {
+            if (serverId == null || serverId.trim().isEmpty()) {


serverId is already trimmed in constructor.

serverId is already trimmed in constructor.

Thanks. Addressed in 6999b9e.

gyang94 · 2026-04-20T03:11:08Z

+
+        private String diskId(File dataDir) throws IOException {
+            if (diskId == null || diskId.trim().isEmpty()) {
+                throw new IOException(


same as above. diskId also was trimmed in the constructor.

same as above. diskId also was trimmed in the constructor.

Thanks. Addressed in 6999b9e.

wuchong

Thanks @hanliu0830 , I left some comments.

wuchong · 2026-05-05T08:35:45Z

-            try {
-                Map<TableBucket, Long> recoveryOffsets = new HashMap<>();
-                for (Map.Entry<TableBucket, LogTablet> entry : currentLogs.entrySet()) {
+    void checkpointRecoveryOffsets(File dataDir) {


Could you refactor this method to collect the LogTablets that belong to the dataDir, and directly call checkpointRecoveryOffsets(File dataDir, List<LogTablet> logs)? This helps us to reuse code.

wuchong · 2026-05-05T09:52:54Z

Code review

Found 1 issue:

throw e.getCause() will throw NullPointerException when the caught exception is an InterruptedException, since InterruptedException.getCause() returns null by default. The InterruptedException and ExecutionException cases should be handled separately — for InterruptedException, restore the interrupt flag and rethrow; for ExecutionException, unwrap with a null check on getCause().

fluss/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java

Lines 180 to 185 in 9bc453c

    
               try { 
        
                   future.get(); 
        
               } catch (InterruptedException | ExecutionException e) { 
        
                   throw e.getCause(); 
        
               } 
        
           }

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

wuchong

Besides, The JBOD multi-directory feature has comprehensive unit tests (LocalDiskManagerTest with 14 cases, ReplicaManagerTest with 3 JBOD-tagged cases, plus adapted LogManagerTest/KvManagerTest), but it lacks integration tests (ITCase) that validate the end-to-end behavior in a real multi-TabletServer cluster. I'd suggest adding a JbodMultiDirITCase under fluss-server/.../server/replica/ (alongside existing KvSnapshotITCase, ReplicaFetcherITCase, etc.) using FlussClusterExtension with data.dirs configured to multiple directories. Key scenarios to cover: (1) creating a table and verifying buckets are distributed across directories, (2) writing and reading data correctly under multi-dir setup, (3) PK table Log/KV co-location across a real cluster. This would require first extending FlussClusterExtension to support per-TabletServer data.dirs configuration if it doesn't already.

hanliu0830 · 2026-05-07T03:51:27Z

Code review

Found 1 issue:

throw e.getCause() will throw NullPointerException when the caught exception is an InterruptedException, since InterruptedException.getCause() returns null by default. The InterruptedException and ExecutionException cases should be handled separately — for InterruptedException, restore the interrupt flag and rethrow; for ExecutionException, unwrap with a null check on getCause().

fluss/fluss-server/src/main/java/org/apache/fluss/server/log/LogManager.java

Lines 180 to 185 in 9bc453c

try {

future.get();

} catch (InterruptedException | ExecutionException e) {

throw e.getCause();

}

}

🤖 Generated with Claude Code

If this code review was useful, please react with 👍. Otherwise, react with 👎.

Addressed in dc334c3.Yes, I split the handling of InterruptedException and ExecutionException

wuchong

I pushed a commit to refactor checkpointRecoveryOffsets(File dataDir) to invoke checkpointRecoveryOffsets(File dataDir, List logs) which is minor improvements. Others looks good to me.

I will merge this once the CI is passed.

The integration IT case JbodMultiDirITCase can be added in a follow up PR.

…RecoveryOffsets(File dataDir, List<LogTablet> logs)

wuchong requested a review from Copilot April 9, 2026 03:07

Copilot started reviewing on behalf of wuchong April 9, 2026 03:08 View session

Copilot AI reviewed Apr 9, 2026

View reviewed changes

gyang94 reviewed Apr 20, 2026

View reviewed changes

hanliu0830 force-pushed the fip-33-jbod-phase1 branch from 6999b9e to 8c7375b Compare April 24, 2026 02:43

wuchong reviewed May 5, 2026

View reviewed changes

Comment thread fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java

wuchong force-pushed the fip-33-jbod-phase1 branch from 6dac12a to 87db556 Compare May 9, 2026 08:53

wuchong approved these changes May 9, 2026

View reviewed changes

hanliu0830 and others added 12 commits May 9, 2026 19:15

[server] Implement JBOD Phase 1 Local Multi-Directory Support

8992c33

flix comment

6655744

[rebase] fix MapUtils

f7fee4d

[CR] Handle InterruptedException and ExecutionException reporting

f3b9c40

[CR] Add tryLock explanation

36f85b0

[CR] Remove duplicate thread pools and rename log loading helpers

8959c89

[CR] Remove redundant clean shutdown marker log

0467890

[CR] Make LogTablet the true source of dataDir

e81d839

[CR] Move remote log cache size out of loop

354a2df

Refactor checkpointRecoveryOffsets(File dataDir) to invoke checkpoint…

040ba01

…RecoveryOffsets(File dataDir, List<LogTablet> logs)

fix checkstyle

d64bb5c

rebase main

45f7b8f

wuchong force-pushed the fip-33-jbod-phase1 branch from 13f66b6 to 45f7b8f Compare May 9, 2026 11:17

fix compile

a0cf65b

wuchong merged commit d4cd1a2 into apache:main May 10, 2026
7 checks passed

	LOG.warn("No disk.properties file under dir {}", file.getAbsolutePath());
	LOG.info("No disk.properties file under dir {}", file.getAbsolutePath());

	private static final String JBOD_MULTI_DIR_TAG = "jbod-multidir";
	private static final String JBOD_MULTI_DIR_TAG =
	org.apache.fluss.server.replica.ReplicaTestBase.JBOD_MULTI_DIR_TAG;

Conversation

hanliu0830 commented Apr 8, 2026

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Apr 9, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 9, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 9, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 9, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gyang94 Apr 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gyang94 Apr 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong commented May 5, 2026

Code review

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hanliu0830 commented May 7, 2026

Code review

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

gyang94 Apr 20, 2026 •

edited

Loading

gyang94 Apr 20, 2026 •

edited

Loading