You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/09/24 15:20:35 UTC
(fluss) branch main updated: [server] Eliminate global lock for KV tablet creation/deletion/load (#3994)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni 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 b530fcc2b [server] Eliminate global lock for KV tablet creation/deletion/load (#3994)
b530fcc2b is described below
commit b530fcc2b73d4e485925003de3752d76f7bd913f
Author: Junfan Zhang <zu...@apache.org>
AuthorDate: Thu Sep 24 23:20:29 2026 +0800
[server] Eliminate global lock for KV tablet creation/deletion/load (#3994)
* [server] Eliminate global lock for KV tablet creation/deletion/load
* [server] Test KV bucket lock removal
---------
Co-authored-by: zhangjunfan <zh...@qiyi.com>
Co-authored-by: Anton Borisov <an...@fresha.com>
---
.../org/apache/fluss/server/TabletManagerBase.java | 4 -
.../java/org/apache/fluss/server/kv/KvManager.java | 162 +++++++++++++--------
.../org/apache/fluss/server/kv/KvManagerTest.java | 144 ++++++++++++++++++
3 files changed, 249 insertions(+), 61 deletions(-)
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 187f2c5a1..2d79893a5 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
@@ -51,8 +51,6 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -81,8 +79,6 @@ public abstract class TabletManagerBase {
protected final Configuration conf;
- protected final Lock tabletCreationOrDeletionLock = new ReentrantLock();
-
// TODO make this parameter configurable.
private final int recoveryThreads;
private final TabletType tabletType;
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 1ac64e786..29eedc5eb 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
@@ -51,6 +51,7 @@ import org.apache.fluss.utils.FlussPaths;
import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.clock.Clock;
import org.apache.fluss.utils.clock.SystemClock;
+import org.apache.fluss.utils.function.SupplierWithException;
import org.apache.fluss.utils.types.Tuple2;
import org.rocksdb.Cache;
@@ -85,7 +86,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Predicate;
import static org.apache.fluss.utils.Preconditions.checkState;
-import static org.apache.fluss.utils.concurrent.LockUtils.inLock;
/**
* The entry point to the fluss kv management subsystem. The kv manager is responsible for kv tablet
@@ -140,6 +140,8 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
private final Map<TableBucket, KvTablet> currentKvs = new ConcurrentHashMap<>();
+ private final Map<TableBucket, Object> kvLocks = new ConcurrentHashMap<>();
+
/**
* For arrow log format. The buffer allocator to allocate memory for arrow write batch of
* changelog records.
@@ -385,46 +387,39 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
}
private void cleanupStaleKvDirectories() {
- inLock(
- tabletCreationOrDeletionLock,
- () -> {
- checkState(!isShutdown, "Cannot clean KV directories after shutdown.");
- checkState(
- currentKvs.isEmpty(),
- "Cannot clean KV directories while KV tablets are open.");
- for (File dataDir : dataDirs) {
- try {
- Path realDataDir = dataDir.toPath().toRealPath();
- List<File> staleDirs =
- listTabletsToLoad(
- realDataDir.toFile(), this::listCleanupDirectories);
- int deletedDirectories = 0;
- for (File tabletDir : staleDirs) {
- try {
- deleteStaleKvDirectory(tabletDir.toPath(), realDataDir);
- deletedDirectories++;
- } catch (IOException e) {
- LOG.warn(
- "Failed to clean stale KV tablet directory {}. "
- + "Continuing cleanup of other tablet directories.",
- tabletDir,
- e);
- }
- }
- LOG.info(
- "Cleaned up {} of {} stale KV tablet directories in {}.",
- deletedDirectories,
- staleDirs.size(),
- dataDir);
- } catch (IOException e) {
- LOG.warn(
- "Failed to clean stale KV directories in {}. Skipping remaining "
- + "cleanup for this data directory; startup will continue.",
- dataDir,
- e);
- }
+ checkState(!isShutdown, "Cannot clean KV directories after shutdown.");
+ checkState(currentKvs.isEmpty(), "Cannot clean KV directories while KV tablets are open.");
+ for (File dataDir : dataDirs) {
+ try {
+ Path realDataDir = dataDir.toPath().toRealPath();
+ List<File> staleDirs =
+ listTabletsToLoad(realDataDir.toFile(), this::listCleanupDirectories);
+ int deletedDirectories = 0;
+ for (File tabletDir : staleDirs) {
+ try {
+ deleteStaleKvDirectory(tabletDir.toPath(), realDataDir);
+ deletedDirectories++;
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to clean stale KV tablet directory {}. "
+ + "Continuing cleanup of other tablet directories.",
+ tabletDir,
+ e);
}
- });
+ }
+ LOG.info(
+ "Cleaned up {} of {} stale KV tablet directories in {}.",
+ deletedDirectories,
+ staleDirs.size(),
+ dataDir);
+ } catch (IOException e) {
+ LOG.warn(
+ "Failed to clean stale KV directories in {}. Skipping remaining "
+ + "cleanup for this data directory; startup will continue.",
+ dataDir,
+ e);
+ }
+ }
}
private void deleteStaleKvDirectory(Path tabletDir, Path dataDir) throws IOException {
@@ -551,8 +546,8 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
ArrowCompressionInfo arrowCompressionInfo,
@Nullable Runnable flushCompleteListener)
throws Exception {
- return inLock(
- tabletCreationOrDeletionLock,
+ return inKvLock(
+ tableBucket,
() -> {
if (currentKvs.containsKey(tableBucket)) {
return currentKvs.get(tableBucket);
@@ -626,9 +621,16 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
}
public void dropKv(TableBucket tableBucket) {
- KvTablet dropKvTablet =
- inLock(tabletCreationOrDeletionLock, () -> currentKvs.remove(tableBucket));
+ inKvLock(
+ tableBucket,
+ () -> {
+ doDropKv(tableBucket);
+ return null;
+ });
+ }
+ private void doDropKv(TableBucket tableBucket) {
+ KvTablet dropKvTablet = currentKvs.remove(tableBucket);
if (dropKvTablet != null) {
TablePath tablePath = dropKvTablet.getTablePath();
try {
@@ -665,6 +667,38 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
Tuple2<PhysicalTablePath, TableBucket> pathAndBucket = FlussPaths.parseTabletDir(tabletDir);
PhysicalTablePath physicalTablePath = pathAndBucket.f0;
TableBucket tableBucket = pathAndBucket.f1;
+ return inKvLock(
+ tableBucket,
+ () ->
+ doLoadKv(
+ tabletDir,
+ physicalTablePath,
+ tableBucket,
+ schemaGetter,
+ flushCompleteListener));
+ }
+
+ private KvTablet doLoadKv(
+ File tabletDir,
+ PhysicalTablePath physicalTablePath,
+ TableBucket tableBucket,
+ SchemaGetter schemaGetter,
+ @Nullable Runnable flushCompleteListener)
+ throws Exception {
+ KvTablet currentKv = currentKvs.get(tableBucket);
+ if (currentKv != null) {
+ throw new IllegalStateException(
+ String.format(
+ "Duplicate kv tablet directories for bucket %s are found in both %s and %s. "
+ + "Recover server from this "
+ + "failure by manually deleting one of the two kv directories for this bucket. "
+ + "It is recommended to delete the bucket in the kv tablet directory that is "
+ + "known to have failed recently.",
+ tableBucket,
+ tabletDir.getAbsolutePath(),
+ currentKv.getKvTabletDir().getAbsolutePath()));
+ }
+
// get the log tablet for the kv tablet
LogTablet logTablet =
logManager
@@ -714,19 +748,7 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
autoIncrementManager,
clock,
tableConfig);
- if (this.currentKvs.containsKey(tableBucket)) {
- throw new IllegalStateException(
- String.format(
- "Duplicate kv tablet directories for bucket %s are found in both %s and %s. "
- + "Recover server from this "
- + "failure by manually deleting one of the two kv directories for this bucket. "
- + "It is recommended to delete the bucket in the kv tablet directory that is "
- + "known to have failed recently.",
- tableBucket,
- tabletDir.getAbsolutePath(),
- currentKvs.get(tableBucket).getKvTabletDir().getAbsolutePath()));
- }
- this.currentKvs.put(tableBucket, kvTablet);
+ currentKvs.put(tableBucket, kvTablet);
return kvTablet;
}
@@ -798,4 +820,30 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
"Failed to reconfigure shared RocksDB rate limiter: " + e.getMessage(), e);
}
}
+
+ @VisibleForTesting
+ boolean hasKvLock(TableBucket tableBucket) {
+ return kvLocks.containsKey(tableBucket);
+ }
+
+ private <T, E extends Exception> T inKvLock(
+ TableBucket tableBucket, SupplierWithException<T, E> action) throws E {
+ while (true) {
+ Object lock = kvLocks.computeIfAbsent(tableBucket, ignored -> new Object());
+ synchronized (lock) {
+ if (kvLocks.get(tableBucket) != lock) {
+ continue;
+ }
+ try {
+ return action.get();
+ } finally {
+ if (!currentKvs.containsKey(tableBucket)) {
+ // Retire the lock only after the action completes, while still holding
+ // its monitor. Threads waiting on this lock must retry with the new one.
+ kvLocks.remove(tableBucket, lock);
+ }
+ }
+ }
+ }
+ }
}
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 31f92f543..c153bfbfa 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
@@ -80,10 +80,12 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.fluss.compression.ArrowCompressionInfo.DEFAULT_COMPRESSION;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK;
@@ -781,6 +783,117 @@ final class KvManagerTest {
assertThat(kvManager.getKv(tableBucket1)).isPresent();
}
+ @Test
+ void testKvRunConcurrentlyForDifferentBuckets() throws Exception {
+ initTableBuckets(null);
+ BlockingSchemaGetter blockingSchemaGetter = new BlockingSchemaGetter();
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<KvTablet> blockedCreation =
+ executor.submit(
+ () -> getOrCreateKv(tablePath1, null, tableBucket1, blockingSchemaGetter));
+ try {
+ blockingSchemaGetter.awaitBlocked();
+
+ Future<KvTablet> otherBucketCreation =
+ executor.submit(() -> getOrCreateKv(tablePath2, null, tableBucket2));
+ assertThat(otherBucketCreation.get(10, TimeUnit.SECONDS)).isNotNull();
+ } finally {
+ blockingSchemaGetter.unblock();
+ blockedCreation.get(10, TimeUnit.SECONDS);
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testDropKvRacingGetOrCreateKv() throws Exception {
+ initTableBuckets(null);
+ KvTablet oldKv = getOrCreateKv(tablePath1, null, tableBucket1);
+ ResourceGuard resourceGuard = oldKv.getRocksDBKv().getResourceGuard();
+ ResourceGuard.Lease lease = resourceGuard.acquireResource();
+ BlockingSchemaGetter blockingSchemaGetter = new BlockingSchemaGetter();
+ ExecutorService executor = Executors.newFixedThreadPool(3);
+ try {
+ Future<?> drop = executor.submit(() -> kvManager.dropKv(tableBucket1));
+ waitUntil(
+ resourceGuard::isClosed,
+ Duration.ofSeconds(10),
+ "The drop should wait for the outstanding lease.");
+
+ AtomicReference<Thread> waitingThread = new AtomicReference<>();
+ Future<KvTablet> waitingCreation =
+ executor.submit(
+ () -> {
+ waitingThread.set(Thread.currentThread());
+ return getOrCreateKv(
+ tablePath1, null, tableBucket1, blockingSchemaGetter);
+ });
+ waitUntil(
+ () ->
+ waitingThread.get() != null
+ && waitingThread.get().getState() == Thread.State.BLOCKED,
+ Duration.ofSeconds(10),
+ "The first creation should wait on the lock held by dropKv.");
+
+ lease.close();
+ drop.get(10, TimeUnit.SECONDS);
+ blockingSchemaGetter.awaitBlocked();
+
+ // This arrival must share the replacement lock with the waiter on the old lock.
+ AtomicReference<Thread> concurrentThread = new AtomicReference<>();
+ Future<KvTablet> concurrentCreation =
+ executor.submit(
+ () -> {
+ concurrentThread.set(Thread.currentThread());
+ return getOrCreateKv(tablePath1, null, tableBucket1);
+ });
+ waitUntil(
+ () ->
+ concurrentCreation.isDone()
+ || (concurrentThread.get() != null
+ && concurrentThread.get().getState()
+ == Thread.State.BLOCKED),
+ Duration.ofSeconds(10),
+ "The second creation should wait on the replacement lock.");
+ assertThat(concurrentCreation.isDone()).isFalse();
+
+ blockingSchemaGetter.unblock();
+ KvTablet recreatedKv = waitingCreation.get(10, TimeUnit.SECONDS);
+ assertThat(recreatedKv).isNotSameAs(oldKv);
+ assertThat(concurrentCreation.get(10, TimeUnit.SECONDS)).isSameAs(recreatedKv);
+ assertThat(kvManager.getKv(tableBucket1)).contains(recreatedKv);
+ } finally {
+ lease.close();
+ blockingSchemaGetter.unblock();
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ void testKvLockRemovedWhenBucketHasNoKv() throws Exception {
+ initTableBuckets(null);
+ getOrCreateKv(tablePath1, null, tableBucket1);
+ assertThat(kvManager.hasKvLock(tableBucket1)).isTrue();
+
+ kvManager.dropKv(tableBucket1);
+ assertThat(kvManager.hasKvLock(tableBucket1)).isFalse();
+
+ kvManager.dropKv(tableBucket2);
+ assertThat(kvManager.hasKvLock(tableBucket2)).isFalse();
+
+ SchemaGetter failingSchemaGetter =
+ new TestingSchemaGetter(new SchemaInfo(DATA1_SCHEMA_PK, 1)) {
+ @Override
+ public SchemaInfo getLatestSchemaInfo() {
+ throw new FlussRuntimeException("Failed schema lookup.");
+ }
+ };
+ assertThatThrownBy(() -> getOrCreateKv(tablePath1, null, tableBucket1, failingSchemaGetter))
+ .hasMessageContaining("Failed schema lookup.");
+ assertThat(kvManager.getKv(tableBucket1)).isNotPresent();
+ assertThat(kvManager.hasKvLock(tableBucket1)).isFalse();
+ }
+
@Test
void testGetNonExistentKv() {
initTableBuckets(null);
@@ -984,4 +1097,35 @@ final class KvManagerTest {
return ((Number) ((Gauge<?>) metricGroup.getMetrics().get(metricName)).getValue())
.longValue();
}
+
+ private static final class BlockingSchemaGetter extends TestingSchemaGetter {
+ private final CountDownLatch blocked = new CountDownLatch(1);
+ private final CountDownLatch unblock = new CountDownLatch(1);
+
+ private BlockingSchemaGetter() {
+ super(new SchemaInfo(DATA1_SCHEMA_PK, 1));
+ }
+
+ @Override
+ public SchemaInfo getLatestSchemaInfo() {
+ blocked.countDown();
+ try {
+ if (!unblock.await(30, TimeUnit.SECONDS)) {
+ throw new FlussRuntimeException("Timed out waiting to unblock schema lookup.");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new FlussRuntimeException("Interrupted while blocking schema lookup.", e);
+ }
+ return super.getLatestSchemaInfo();
+ }
+
+ private void awaitBlocked() throws InterruptedException {
+ assertThat(blocked.await(10, TimeUnit.SECONDS)).isTrue();
+ }
+
+ private void unblock() {
+ unblock.countDown();
+ }
+ }
}