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 6320400ff [server] Reject KV leader promotion under disk write lock
(#3539)
6320400ff is described below
commit 6320400ff662ae36ac76fbb755748d246b44ea28
Author: yunhong <[email protected]>
AuthorDate: Tue Jul 7 10:15:25 2026 +0800
[server] Reject KV leader promotion under disk write lock (#3539)
---
.../org/apache/fluss/config/ConfigOptions.java | 11 +
.../server/coordinator/AutoPartitionManager.java | 2 +
.../server/coordinator/CoordinatorContext.java | 34 +++
.../coordinator/CoordinatorEventProcessor.java | 109 +++++++++
.../server/coordinator/CoordinatorServer.java | 22 ++
.../coordinator/LakeTableTieringManager.java | 2 +
.../coordinator/TableLifecycleThrottler.java | 2 +
.../coordinator/event/RetryOfflineLeaderEvent.java | 21 ++
.../coordinator/lease/KvSnapshotLeaseManager.java | 3 +
.../producer/ProducerOffsetsManager.java | 2 +
.../coordinator/rebalance/RebalanceManager.java | 2 +
.../fluss/server/replica/ReplicaManager.java | 17 +-
.../server/coordinator/CoordinatorContextTest.java | 44 ++++
.../coordinator/CoordinatorEventProcessorTest.java | 270 ++++++++++++++++++++-
.../rebalance/RebalanceManagerTest.java | 10 +
.../statemachine/TableBucketStateMachineTest.java | 15 ++
.../fluss/server/replica/ReplicaManagerTest.java | 58 +++++
website/docs/maintenance/configuration.md | 13 +-
18 files changed, 628 insertions(+), 9 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 aca5edaa8..c34c3626f 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
@@ -221,6 +221,17 @@ public class ConfigOptions {
+ "TableLifecycleThrottler scans
in-flight drops for "
+ "timeouts.");
+ public static final ConfigOption<Duration>
COORDINATOR_OFFLINE_LEADER_RETRY_DELAY =
+ key("coordinator.offline-leader.retry-delay")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(1))
+ .withDescription(
+ "The delay before the coordinator retries offline
leaders on live "
+ + "tablet servers after they are marked
offline. This lets a "
+ + "leader that was rejected because of
temporary tablet-server "
+ + "conditions, such as disk write
protection, become electable "
+ + "again after recovery.");
+
public static final ConfigOption<Boolean> LOG_TABLE_ALLOW_CREATION =
key("allow.create.log.tables")
.booleanType()
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
index 7ac0b2fcf..0a817c0e1 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/AutoPartitionManager.java
@@ -122,6 +122,8 @@ public class AutoPartitionManager implements AutoCloseable {
remoteDirDynamicLoader,
conf,
SystemClock.getInstance(),
+ // TODO: Reuse the CoordinatorServer shared scheduler for this
lightweight
+ // coordinator periodic task instead of creating a
component-owned scheduler.
Executors.newScheduledThreadPool(
1, new
ExecutorThreadFactory("periodic-auto-partition-manager")));
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
index dc0fa58a3..6a5710ad3 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
@@ -225,6 +225,40 @@ public class CoordinatorContext {
replicasOnOffline.remove(serverId);
}
+ /** Removes the offline marker for one table bucket on the given tablet
server. */
+ public void removeOfflineBucketInServer(TableBucket tableBucket, int
serverId) {
+ Set<TableBucket> tableBuckets = replicasOnOffline.get(serverId);
+ if (tableBuckets == null) {
+ return;
+ }
+ tableBuckets.remove(tableBucket);
+ if (tableBuckets.isEmpty()) {
+ replicasOnOffline.remove(serverId);
+ }
+ }
+
+ /**
+ * Returns offline replicas that are on live tablet servers.
+ *
+ * <p>The {@code replicasOnOffline} map is part of {@link
#isReplicaOnline(int, TableBucket)},
+ * so replicas in it are filtered out from leader election even when their
tablet server is
+ * still live.
+ */
+ public Set<TableBucketReplica> offlineReplicasOnLiveTabletServers() {
+ Set<Integer> liveTabletServers = liveTabletServerSet();
+ Set<TableBucketReplica> offlineReplicas = new HashSet<>();
+ for (Map.Entry<Integer, Set<TableBucket>> entry :
replicasOnOffline.entrySet()) {
+ int serverId = entry.getKey();
+ if (!liveTabletServers.contains(serverId)) {
+ continue;
+ }
+ for (TableBucket tableBucket : entry.getValue()) {
+ offlineReplicas.add(new TableBucketReplica(tableBucket,
serverId));
+ }
+ }
+ return offlineReplicas;
+ }
+
// ---- Pending leader activation tracking (for Cluster Health API) ----
public void addPendingLeaderActivation(TableBucket bucket) {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 537e2df63..803b657f4 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -88,6 +88,7 @@ import
org.apache.fluss.server.coordinator.event.RebalanceEvent;
import org.apache.fluss.server.coordinator.event.RebalanceTaskTimeoutEvent;
import org.apache.fluss.server.coordinator.event.RemoveServerTagEvent;
import org.apache.fluss.server.coordinator.event.ResumeDropEvent;
+import org.apache.fluss.server.coordinator.event.RetryOfflineLeaderEvent;
import org.apache.fluss.server.coordinator.event.SchemaChangeEvent;
import org.apache.fluss.server.coordinator.event.TableRegistrationChangeEvent;
import
org.apache.fluss.server.coordinator.event.watcher.CoordinatorChangeWatcher;
@@ -126,6 +127,7 @@ import
org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
import org.apache.fluss.utils.AutoPartitionStrategy;
import org.apache.fluss.utils.clock.Clock;
import org.apache.fluss.utils.clock.SystemClock;
+import org.apache.fluss.utils.concurrent.Scheduler;
import org.apache.fluss.utils.types.Tuple2;
import org.slf4j.Logger;
@@ -146,6 +148,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors;
import static
org.apache.fluss.server.coordinator.statemachine.BucketState.OfflineBucket;
@@ -188,8 +191,11 @@ public class CoordinatorEventProcessor implements
EventProcessor {
private final String internalListenerName;
private final CoordinatorMetricGroup coordinatorMetricGroup;
private final RebalanceManager rebalanceManager;
+ private final Scheduler scheduler;
+ private final long offlineLeaderRetryDelayMs;
private final CompletedSnapshotStoreManager completedSnapshotStoreManager;
private final LakeTableHelper lakeTableHelper;
+ private ScheduledFuture<?> offlineLeaderRetryTask;
public CoordinatorEventProcessor(
ZooKeeperClient zooKeeperClient,
@@ -203,6 +209,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
ExecutorService ioExecutor,
MetadataManager metadataManager,
KvSnapshotLeaseManager kvSnapshotLeaseManager,
+ Scheduler scheduler,
Clock clock) {
this.zooKeeperClient = zooKeeperClient;
this.serverMetadataCache = serverMetadataCache;
@@ -262,6 +269,16 @@ public class CoordinatorEventProcessor implements
EventProcessor {
this.rebalanceManager =
new RebalanceManager(
this, zooKeeperClient, coordinatorEventManager,
SystemClock.getInstance());
+ this.offlineLeaderRetryDelayMs =
+
conf.get(ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY).toMillis();
+ if (offlineLeaderRetryDelayMs <= 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "%s must be positive, but was %d ms.",
+
ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY.key(),
+ offlineLeaderRetryDelayMs));
+ }
+ this.scheduler = scheduler;
this.ioExecutor = ioExecutor;
this.lakeTableHelper =
new LakeTableHelper(zooKeeperClient,
conf.getString(ConfigOptions.REMOTE_DATA_DIR));
@@ -284,6 +301,11 @@ public class CoordinatorEventProcessor implements
EventProcessor {
return lifecycleThrottler;
}
+ @VisibleForTesting
+ boolean hasOfflineLeaderRetryTaskScheduled() {
+ return offlineLeaderRetryTask != null;
+ }
+
public void startup() {
coordinatorContext.setCoordinatorServerInfo(getCoordinatorServerInfo());
// start watchers first so that we won't miss node in zk;
@@ -325,6 +347,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
}
public void shutdown() {
+ clearOfflineLeaderRetryTask();
// close the event manager
coordinatorEventManager.close();
rebalanceManager.close();
@@ -587,6 +610,39 @@ public class CoordinatorEventProcessor implements
EventProcessor {
tabletServerChangeWatcher.stop();
}
+ private void scheduleOfflineLeaderRetryIfNeeded() {
+ if (offlineLeaderRetryTask != null) {
+ return;
+ }
+
+ // Keep CoordinatorContext access on the coordinator event thread. The
scheduled task only
+ // enqueues a retry event.
+ Set<TableBucketReplica> retryableOfflineReplicas =
+ retryableOfflineReplicasOnLiveTabletServers();
+ if (retryableOfflineReplicas.isEmpty()) {
+ return;
+ }
+
+ offlineLeaderRetryTask =
+ scheduler.scheduleOnce(
+ "offline-leader-retry",
+ this::enqueueRetryOfflineLeaderEventSafely,
+ offlineLeaderRetryDelayMs);
+ LOG.info(
+ "Offline leader retry task scheduled after {} ms for {}
replicas: {}.",
+ offlineLeaderRetryDelayMs,
+ retryableOfflineReplicas.size(),
+ retryableOfflineReplicas);
+ }
+
+ private void enqueueRetryOfflineLeaderEventSafely() {
+ try {
+ coordinatorEventManager.put(new RetryOfflineLeaderEvent());
+ } catch (Throwable t) {
+ LOG.warn("Failed to enqueue retry offline leader event.", t);
+ }
+ }
+
@Override
public void process(CoordinatorEvent event) {
if (event instanceof CreateTableEvent) {
@@ -605,6 +661,8 @@ public class CoordinatorEventProcessor implements
EventProcessor {
} else if (event instanceof NotifyLeaderAndIsrResponseReceivedEvent) {
processNotifyLeaderAndIsrResponseReceivedEvent(
(NotifyLeaderAndIsrResponseReceivedEvent) event);
+ } else if (event instanceof RetryOfflineLeaderEvent) {
+ processRetryOfflineLeader();
} else if (event instanceof DeleteReplicaResponseReceivedEvent) {
processDeleteReplicaResponseReceived((DeleteReplicaResponseReceivedEvent)
event);
} else if (event instanceof NewCoordinatorEvent) {
@@ -702,6 +760,56 @@ public class CoordinatorEventProcessor implements
EventProcessor {
}
}
+ private void processRetryOfflineLeader() {
+ // This event consumes the currently scheduled one-shot retry. If the
NotifyLeaderAndIsr
+ // request sent below is rejected by the target tablet server again,
its response handler
+ // will call onReplicaBecomeOffline(), put the offline marker back,
and schedule the next
+ // one-shot retry.
+ clearOfflineLeaderRetryTask();
+ Set<TableBucketReplica> offlineReplicas =
retryableOfflineReplicasOnLiveTabletServers();
+
+ if (offlineReplicas.isEmpty()) {
+ return;
+ }
+
+ LOG.info(
+ "Retrying {} offline replicas on live tablet servers before
triggering "
+ + "offline leader election: {}.",
+ offlineReplicas.size(),
+ offlineReplicas);
+
+ // replicasOnOffline is checked by CoordinatorContext#isReplicaOnline
and therefore also
+ // by leader election. Remove the marker before probing the live
server again; if the
+ // server still cannot become leader, NotifyLeaderAndIsr will fail and
the replica will be
+ // put back to OfflineReplica. TODO: once standby replicas maintain
local KV snapshots,
+ // distinguish standby promotion from fresh snapshot download when
retrying KV leaders.
+ for (TableBucketReplica offlineReplica : offlineReplicas) {
+ coordinatorContext.removeOfflineBucketInServer(
+ offlineReplica.getTableBucket(),
offlineReplica.getReplica());
+ }
+
+ replicaStateMachine.handleStateChanges(offlineReplicas, OnlineReplica);
+ tableBucketStateMachine.triggerOnlineBucketStateChange();
+ }
+
+ private Set<TableBucketReplica>
retryableOfflineReplicasOnLiveTabletServers() {
+ // offlineReplicasOnLiveTabletServers() returns replicas from
replicasOnOffline, which is
+ // only a leader-election exclusion marker. The replica state may have
changed after the
+ // marker was added, for example because deletion or migration
started. Retry only replicas
+ // that are still OfflineReplica in the state machine.
+ return coordinatorContext.offlineReplicasOnLiveTabletServers().stream()
+ .filter(replica ->
!coordinatorContext.isToBeDeleted(replica.getTableBucket()))
+ .filter(replica -> coordinatorContext.getReplicaState(replica)
== OfflineReplica)
+ .collect(Collectors.toSet());
+ }
+
+ private void clearOfflineLeaderRetryTask() {
+ if (offlineLeaderRetryTask != null) {
+ offlineLeaderRetryTask.cancel(false);
+ offlineLeaderRetryTask = null;
+ }
+ }
+
private void processCreateTable(CreateTableEvent createTableEvent) {
long tableId = createTableEvent.getTableInfo().getTableId();
// skip the table if it already exists
@@ -1114,6 +1222,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
// kafka, todo: but we may need to select another tablet server to put
// replica
replicaStateMachine.handleStateChanges(offlineReplicas,
OfflineReplica);
+ scheduleOfflineLeaderRetryIfNeeded();
}
private void processNewCoordinator(NewCoordinatorEvent
newCoordinatorEvent) {
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
index 2c4fcf5e4..549eaf657 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
@@ -51,7 +51,9 @@ import org.apache.fluss.utils.ExecutorUtils;
import org.apache.fluss.utils.clock.Clock;
import org.apache.fluss.utils.clock.SystemClock;
import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;
+import org.apache.fluss.utils.concurrent.FlussScheduler;
import org.apache.fluss.utils.concurrent.FutureUtils;
+import org.apache.fluss.utils.concurrent.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,6 +71,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.apache.fluss.config.ConfigOptions.BACKGROUND_THREADS;
import static
org.apache.fluss.config.FlussConfigUtils.validateCoordinatorConfigs;
/**
@@ -136,6 +139,10 @@ public class CoordinatorServer extends ServerBase {
@GuardedBy("lock")
private LakeTableTieringManager lakeTableTieringManager;
+ /** Shared scheduler for lightweight coordinator background tasks. */
+ @GuardedBy("lock")
+ private Scheduler scheduler;
+
@GuardedBy("lock")
private ExecutorService ioExecutor;
@@ -214,6 +221,9 @@ public class CoordinatorServer extends ServerBase {
LOG.info("Initializing Coordinator services as standby.");
List<Endpoint> endpoints = Endpoint.loadBindEndpoints(conf,
ServerType.COORDINATOR);
+ this.scheduler = new FlussScheduler(conf.get(BACKGROUND_THREADS));
+ scheduler.startup();
+
// for metrics
this.metricRegistry = MetricRegistry.create(conf, pluginManager);
this.serverMetricGroup =
@@ -337,6 +347,7 @@ public class CoordinatorServer extends ServerBase {
ioExecutor,
metadataManager,
kvSnapshotLeaseManager,
+ scheduler,
clock);
coordinatorEventProcessor.startup();
@@ -539,6 +550,17 @@ public class CoordinatorServer extends ServerBase {
synchronized (lock) {
Throwable exception = null;
+ try {
+ // We must shut down the scheduler early because otherwise,
the scheduler could
+ // touch other resources that might have been shutdown and
cause exceptions.
+ if (scheduler != null) {
+ scheduler.shutdown();
+ scheduler = null;
+ }
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+
try {
if (serverMetricGroup != null) {
serverMetricGroup.close();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
index 60589a999..e281381de 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
@@ -158,6 +158,8 @@ public class LakeTableTieringManager implements
AutoCloseable {
public LakeTableTieringManager(LakeTieringMetricGroup
lakeTieringMetricGroup) {
this(
new DefaultTimer("delay lake tiering", 1_000, 20),
+ // TODO: Reuse the CoordinatorServer shared scheduler for this
lightweight
+ // coordinator timeout checker instead of creating a
component-owned scheduler.
Executors.newSingleThreadScheduledExecutor(
new
ExecutorThreadFactory("fluss-lake-tiering-timeout-checker")),
SystemClock.getInstance(),
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableLifecycleThrottler.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableLifecycleThrottler.java
index d7b6a84dd..00e4c8e0d 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableLifecycleThrottler.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TableLifecycleThrottler.java
@@ -107,6 +107,8 @@ public class TableLifecycleThrottler implements
AutoCloseable {
this(
eventManager,
clock,
+ // TODO: Reuse the CoordinatorServer shared scheduler for this
lightweight
+ // coordinator timeout checker instead of creating a
component-owned scheduler.
Executors.newScheduledThreadPool(
1, new
ExecutorThreadFactory("lifecycle-throttler-timeout")),
conf.get(ConfigOptions.COORDINATOR_LIFECYCLE_THROTTLER_INFLIGHT_TIMEOUT).toMillis(),
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/RetryOfflineLeaderEvent.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/RetryOfflineLeaderEvent.java
new file mode 100644
index 000000000..89d8d2858
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/RetryOfflineLeaderEvent.java
@@ -0,0 +1,21 @@
+/*
+ * 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.coordinator.event;
+
+/** An event for retrying offline leaders on live tablet servers. */
+public class RetryOfflineLeaderEvent implements CoordinatorEvent {}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java
index 6c79a668e..99a1c28b1 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/lease/KvSnapshotLeaseManager.java
@@ -92,6 +92,9 @@ public class KvSnapshotLeaseManager {
leaseExpirationCheckInterval,
zkClient,
remoteDataDir,
+ // TODO: Reuse the CoordinatorServer shared scheduler for this
lightweight
+ // coordinator lease expiration task instead of creating a
component-owned
+ // scheduler.
Executors.newScheduledThreadPool(
1, new
ExecutorThreadFactory("kv-snapshot-lease-cleaner")),
clock,
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManager.java
index e40fcb24c..68a69d527 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManager.java
@@ -88,6 +88,8 @@ public class ProducerOffsetsManager implements AutoCloseable {
this.offsetsStore = offsetsStore;
this.defaultTtlMs = defaultTtlMs;
this.cleanupIntervalMs = cleanupIntervalMs;
+ // TODO: Reuse the CoordinatorServer shared scheduler for this
lightweight coordinator
+ // cleanup task instead of creating a component-owned scheduler.
this.cleanupScheduler =
Executors.newSingleThreadScheduledExecutor(
new
ExecutorThreadFactory("producer-snapshot-cleanup"));
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java
index 9f2f350bd..cc29b982b 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java
@@ -131,6 +131,8 @@ public class RebalanceManager {
zkClient,
eventManager,
clock,
+ // TODO: Reuse the CoordinatorServer shared scheduler for this
lightweight
+ // coordinator timeout checker instead of creating a
component-owned scheduler.
Executors.newScheduledThreadPool(
1, new ExecutorThreadFactory("rebalance-timeout")));
}
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 280e34124..29adf341d 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
@@ -1824,7 +1824,11 @@ public class ReplicaManager implements
ServerReconfigurable {
if (requestLeaderEpoch >= currentLeaderEpoch) {
if (data.getReplicas().contains(serverId)) {
int leaderId = data.getLeader();
- return leaderId == serverId;
+ boolean becomeLeader = leaderId == serverId;
+ if (becomeLeader) {
+ ensureWritableForNewKvLeader(replica, requestLeaderEpoch);
+ }
+ return becomeLeader;
} else {
String errorMessage =
String.format(
@@ -1845,6 +1849,17 @@ public class ReplicaManager implements
ServerReconfigurable {
}
}
+ private void ensureWritableForNewKvLeader(Replica replica, int
requestLeaderEpoch) {
+ if (!replica.isKvTable() || requestLeaderEpoch <=
replica.getLeaderEpoch()) {
+ return;
+ }
+
+ // TODO: Once standby replicas maintain local KV snapshots, allow
promoting a standby
+ // replica while disk write protection is active because it should not
need to download a
+ // large remote snapshot during make-leader.
+ localDiskManager.ensureWritable();
+ }
+
/**
* If all the following conditions are true, we need to put a delayed
write operation into the
* delayed write manager and wait for replication to complete.
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
index b69fe5fa3..557d75a6a 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
@@ -22,6 +22,7 @@ import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableBucketReplica;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
@@ -78,6 +79,41 @@ class CoordinatorContextTest {
assertThat(context.getLakeTableCount()).isEqualTo(2);
}
+ @Test
+ void testOfflineReplicasOnLiveTabletServersOnlyReturnsLiveServers() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket liveBucket = new TableBucket(1L, 0);
+ TableBucket deadBucket = new TableBucket(1L, 1);
+
+ context.setLiveTabletServers(Arrays.asList(createTabletServer(0),
createTabletServer(1)));
+ context.addOfflineBucketInServer(liveBucket, 0);
+ context.addOfflineBucketInServer(deadBucket, 2);
+
+ assertThat(context.offlineReplicasOnLiveTabletServers())
+ .containsExactly(new TableBucketReplica(liveBucket, 0));
+ }
+
+ @Test
+ void testRemoveOfflineBucketInServerForBucket() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket tb1 = new TableBucket(1L, 0);
+ TableBucket tb2 = new TableBucket(1L, 1);
+
+
context.setLiveTabletServers(Collections.singletonList(createTabletServer(0)));
+ context.addOfflineBucketInServer(tb1, 0);
+ context.addOfflineBucketInServer(tb2, 0);
+
+ context.removeOfflineBucketInServer(tb1, 0);
+ assertThat(context.isReplicaOnline(0, tb1)).isTrue();
+ assertThat(context.isReplicaOnline(0, tb2)).isFalse();
+ assertThat(context.offlineReplicasOnLiveTabletServers())
+ .containsExactly(new TableBucketReplica(tb2, 0));
+
+ context.removeOfflineBucketInServer(tb2, 0);
+ assertThat(context.isReplicaOnline(0, tb2)).isTrue();
+ assertThat(context.offlineReplicasOnLiveTabletServers()).isEmpty();
+ }
+
// ---- Pending Leader Activation Tracking Tests ----
@Test
@@ -215,4 +251,12 @@ class CoordinatorContextTest {
System.currentTimeMillis(),
System.currentTimeMillis());
}
+
+ private ServerInfo createTabletServer(int serverId) {
+ return new ServerInfo(
+ serverId,
+ "RACK" + serverId,
+ Endpoint.fromListenersString("CLIENT://host" + serverId +
":9124"),
+ ServerType.TABLET_SERVER);
+ }
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
index 001a90f74..d92953cc2 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
@@ -46,12 +46,16 @@ import
org.apache.fluss.rpc.messages.NotifyLeaderAndIsrRequest;
import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse;
import org.apache.fluss.rpc.messages.NotifyRemoteLogOffsetsRequest;
import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
+import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.rpc.protocol.Errors;
import org.apache.fluss.server.coordinator.event.AccessContextEvent;
import org.apache.fluss.server.coordinator.event.AdjustIsrReceivedEvent;
import org.apache.fluss.server.coordinator.event.CommitKvSnapshotEvent;
import org.apache.fluss.server.coordinator.event.CommitRemoteLogManifestEvent;
import org.apache.fluss.server.coordinator.event.CoordinatorEventManager;
+import
org.apache.fluss.server.coordinator.event.NotifyLeaderAndIsrResponseReceivedEvent;
+import org.apache.fluss.server.coordinator.event.RetryOfflineLeaderEvent;
import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager;
import org.apache.fluss.server.coordinator.remote.RemoteDirDynamicLoader;
import org.apache.fluss.server.coordinator.statemachine.BucketState;
@@ -59,6 +63,7 @@ import
org.apache.fluss.server.coordinator.statemachine.ReplicaState;
import org.apache.fluss.server.entity.AdjustIsrResultForBucket;
import org.apache.fluss.server.entity.CommitKvSnapshotData;
import org.apache.fluss.server.entity.CommitRemoteLogManifestData;
+import org.apache.fluss.server.entity.NotifyLeaderAndIsrResultForBucket;
import org.apache.fluss.server.entity.TablePropertyChanges;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
import
org.apache.fluss.server.kv.snapshot.ZooKeeperCompletedSnapshotHandleStore;
@@ -87,6 +92,8 @@ import org.apache.fluss.types.DataTypes;
import org.apache.fluss.utils.ExceptionUtils;
import org.apache.fluss.utils.clock.SystemClock;
import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;
+import org.apache.fluss.utils.concurrent.FlussScheduler;
+import org.apache.fluss.utils.concurrent.Scheduler;
import org.apache.fluss.utils.types.Tuple2;
import org.junit.jupiter.api.AfterEach;
@@ -109,6 +116,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -165,6 +173,7 @@ class CoordinatorEventProcessorTest {
private CompletedSnapshotStoreManager completedSnapshotStoreManager;
private CoordinatorMetadataCache serverMetadataCache;
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
+ private Scheduler scheduler;
private String remoteDataDir;
@BeforeAll
@@ -222,6 +231,9 @@ class CoordinatorEventProcessorTest {
TestingMetricGroups.COORDINATOR_METRICS);
kvSnapshotLeaseManager.start();
+ scheduler = new FlussScheduler(1);
+ scheduler.startup();
+
eventProcessor = buildCoordinatorEventProcessor();
eventProcessor.startup();
metadataManager.createDatabase(
@@ -230,8 +242,13 @@ class CoordinatorEventProcessorTest {
}
@AfterEach
- void afterEach() {
- eventProcessor.shutdown();
+ void afterEach() throws Exception {
+ if (eventProcessor != null) {
+ eventProcessor.shutdown();
+ }
+ if (scheduler != null) {
+ scheduler.shutdown();
+ }
metadataManager.dropDatabase(defaultDatabase, false, true);
// clear the assignment info for all tables;
ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupPath(TableIdsZNode.path());
@@ -1112,6 +1129,234 @@ class CoordinatorEventProcessorTest {
assertThat(resultForBucketMap.values()).allMatch(AdjustIsrResultForBucket::succeeded);
}
+ @Test
+ void testDiskWriteLockedNotifyLeaderResponseMarksReplicaOffline() throws
Exception {
+ initCoordinatorChannel();
+ TablePath tablePath =
+ TablePath.of(defaultDatabase,
"disk_write_locked_notify_leader_response");
+ int nBuckets = 3;
+ int replicationFactor = 3;
+ TableAssignment tableAssignment =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {
+ new TabletServerInfo(0, "rack0"),
+ new TabletServerInfo(1, "rack1"),
+ new TabletServerInfo(2, "rack2")
+ });
+ long tableId =
+ metadataManager.createTable(
+ tablePath, remoteDataDir, TEST_TABLE, tableAssignment,
false);
+ verifyTableCreated(tableId, tableAssignment, nBuckets,
replicationFactor);
+
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ int leader =
+
tableAssignment.getBucketAssignment(tableBucket.getBucket()).getReplicas().get(0);
+ TableBucketReplica tableBucketReplica = new
TableBucketReplica(tableBucket, leader);
+
+ eventProcessor
+ .getCoordinatorEventManager()
+ .put(
+ new NotifyLeaderAndIsrResponseReceivedEvent(
+ Collections.singletonList(
+ new NotifyLeaderAndIsrResultForBucket(
+ tableBucket,
+ new ApiError(
+
Errors.DISK_WRITE_LOCKED,
+ "disk write locked"))),
+ leader));
+
+ fromCtx(
+ ctx -> {
+
assertThat(ctx.getReplicaState(tableBucketReplica)).isEqualTo(OfflineReplica);
+ assertThat(ctx.isReplicaOnline(leader,
tableBucket)).isFalse();
+ return null;
+ });
+ }
+
+ @Test
+ void testRetryOfflineLeaderEventRetriesOfflineReplicaOnLiveServer() throws
Exception {
+
assertThat(eventProcessor.hasOfflineLeaderRetryTaskScheduled()).isFalse();
+ initCoordinatorChannel();
+ TablePath tablePath = TablePath.of(defaultDatabase,
"retry_offline_leader_on_live_server");
+ int nBuckets = 3;
+ int replicationFactor = 1;
+ TableAssignment tableAssignment =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {new TabletServerInfo(0,
"rack0")});
+ long tableId =
+ metadataManager.createTable(
+ tablePath,
+ remoteDataDir,
+ TEST_TABLE.withReplicationFactor(replicationFactor),
+ tableAssignment,
+ false);
+ verifyTableCreated(tableId, tableAssignment, nBuckets,
replicationFactor);
+
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ int leader =
+
tableAssignment.getBucketAssignment(tableBucket.getBucket()).getReplicas().get(0);
+ TableBucketReplica tableBucketReplica = new
TableBucketReplica(tableBucket, leader);
+
+ eventProcessor
+ .getCoordinatorEventManager()
+ .put(
+ new NotifyLeaderAndIsrResponseReceivedEvent(
+ Collections.singletonList(
+ new NotifyLeaderAndIsrResultForBucket(
+ tableBucket,
+ new ApiError(
+
Errors.DISK_WRITE_LOCKED,
+ "disk write locked"))),
+ leader));
+
+ retryVerifyContext(
+ ctx -> {
+
assertThat(ctx.getReplicaState(tableBucketReplica)).isEqualTo(OfflineReplica);
+
assertThat(ctx.getBucketState(tableBucket)).isEqualTo(OfflineBucket);
+ assertThat(ctx.isReplicaOnline(leader,
tableBucket)).isFalse();
+ });
+
assertThat(eventProcessor.hasOfflineLeaderRetryTaskScheduled()).isTrue();
+
+ eventProcessor.getCoordinatorEventManager().put(new
RetryOfflineLeaderEvent());
+
+ retryVerifyContext(
+ ctx -> {
+ assertThat(ctx.isReplicaOnline(leader,
tableBucket)).isTrue();
+
assertThat(ctx.getReplicaState(tableBucketReplica)).isEqualTo(OnlineReplica);
+
assertThat(ctx.getBucketState(tableBucket)).isEqualTo(OnlineBucket);
+
assertThat(ctx.getBucketLeaderAndIsr(tableBucket).get().leader())
+ .isEqualTo(leader);
+ });
+
assertThat(eventProcessor.hasOfflineLeaderRetryTaskScheduled()).isFalse();
+ }
+
+ @Test
+ void testRetryOfflineLeaderEventKeepsReplicaOfflineWhenRetryStillFails()
throws Exception {
+ initCoordinatorChannel();
+ TablePath tablePath = TablePath.of(defaultDatabase,
"retry_offline_leader_still_fails");
+ int nBuckets = 3;
+ int replicationFactor = 1;
+ TableAssignment tableAssignment =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {new TabletServerInfo(0,
"rack0")});
+ long tableId =
+ metadataManager.createTable(
+ tablePath,
+ remoteDataDir,
+ TEST_TABLE.withReplicationFactor(replicationFactor),
+ tableAssignment,
+ false);
+ verifyTableCreated(tableId, tableAssignment, nBuckets,
replicationFactor);
+
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ int leader =
+
tableAssignment.getBucketAssignment(tableBucket.getBucket()).getReplicas().get(0);
+ TableBucketReplica tableBucketReplica = new
TableBucketReplica(tableBucket, leader);
+
+ eventProcessor
+ .getCoordinatorEventManager()
+ .put(
+ new NotifyLeaderAndIsrResponseReceivedEvent(
+ Collections.singletonList(
+ new NotifyLeaderAndIsrResultForBucket(
+ tableBucket,
+ new ApiError(
+
Errors.DISK_WRITE_LOCKED,
+ "disk write locked"))),
+ leader));
+
+ retryVerifyContext(
+ ctx -> {
+
assertThat(ctx.getReplicaState(tableBucketReplica)).isEqualTo(OfflineReplica);
+ assertThat(ctx.isReplicaOnline(leader,
tableBucket)).isFalse();
+ });
+
assertThat(eventProcessor.hasOfflineLeaderRetryTaskScheduled()).isTrue();
+
+ CountingFailingNotifyGateway failingGateway = new
CountingFailingNotifyGateway();
+
testCoordinatorChannelManager.setGateways(Collections.singletonMap(leader,
failingGateway));
+
+ eventProcessor.getCoordinatorEventManager().put(new
RetryOfflineLeaderEvent());
+
+ retry(
+ Duration.ofMinutes(1),
+ () ->
assertThat(failingGateway.getNotifyLeaderAndIsrCount()).isGreaterThan(0));
+ retryVerifyContext(
+ ctx -> {
+
assertThat(ctx.getReplicaState(tableBucketReplica)).isEqualTo(OfflineReplica);
+
assertThat(ctx.getBucketState(tableBucket)).isEqualTo(OfflineBucket);
+ assertThat(ctx.isReplicaOnline(leader,
tableBucket)).isFalse();
+ });
+
assertThat(eventProcessor.hasOfflineLeaderRetryTaskScheduled()).isTrue();
+ }
+
+ @Test
+ void testRetryOfflineLeaderEventSkipsDeletedBucket() throws Exception {
+ initCoordinatorChannel();
+ TablePath tablePath = TablePath.of(defaultDatabase,
"retry_offline_leader_deleted_bucket");
+ int nBuckets = 3;
+ int replicationFactor = 1;
+ TableAssignment tableAssignment =
+ generateAssignment(
+ nBuckets,
+ replicationFactor,
+ new TabletServerInfo[] {new TabletServerInfo(0,
"rack0")});
+ long tableId =
+ metadataManager.createTable(
+ tablePath,
+ remoteDataDir,
+ TEST_TABLE.withReplicationFactor(replicationFactor),
+ tableAssignment,
+ false);
+ verifyTableCreated(tableId, tableAssignment, nBuckets,
replicationFactor);
+
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ int leader =
+
tableAssignment.getBucketAssignment(tableBucket.getBucket()).getReplicas().get(0);
+ TableBucketReplica tableBucketReplica = new
TableBucketReplica(tableBucket, leader);
+
+ eventProcessor
+ .getCoordinatorEventManager()
+ .put(
+ new NotifyLeaderAndIsrResponseReceivedEvent(
+ Collections.singletonList(
+ new NotifyLeaderAndIsrResultForBucket(
+ tableBucket,
+ new ApiError(
+
Errors.DISK_WRITE_LOCKED,
+ "disk write locked"))),
+ leader));
+
+ retryVerifyContext(
+ ctx -> {
+
assertThat(ctx.getReplicaState(tableBucketReplica)).isEqualTo(OfflineReplica);
+ assertThat(ctx.isReplicaOnline(leader,
tableBucket)).isFalse();
+ });
+
+ fromCtx(
+ ctx -> {
+ ctx.queueTableDeletion(Collections.singleton(tableId));
+ return null;
+ });
+
+ eventProcessor.getCoordinatorEventManager().put(new
RetryOfflineLeaderEvent());
+
+ fromCtx(
+ ctx -> {
+
assertThat(ctx.getReplicaState(tableBucketReplica)).isEqualTo(OfflineReplica);
+ assertThat(ctx.isReplicaOnline(leader,
tableBucket)).isFalse();
+ assertThat(ctx.offlineReplicasOnLiveTabletServers())
+ .contains(tableBucketReplica);
+ return null;
+ });
+
assertThat(eventProcessor.hasOfflineLeaderRetryTaskScheduled()).isFalse();
+ }
+
@Test
void testSchemaChange() throws Exception {
// make sure all request to gateway should be successful
@@ -1604,6 +1849,7 @@ class CoordinatorEventProcessorTest {
private CoordinatorEventProcessor buildCoordinatorEventProcessor() {
Configuration conf = new Configuration();
conf.set(ConfigOptions.REMOTE_DATA_DIR, remoteDataDir);
+ conf.set(ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY,
Duration.ofDays(1));
return new CoordinatorEventProcessor(
zookeeperClient,
serverMetadataCache,
@@ -1616,6 +1862,7 @@ class CoordinatorEventProcessorTest {
Executors.newFixedThreadPool(1, new
ExecutorThreadFactory("test-coordinator-io")),
metadataManager,
kvSnapshotLeaseManager,
+ scheduler,
SystemClock.getInstance());
}
@@ -2001,6 +2248,25 @@ class CoordinatorEventProcessorTest {
return count;
}
+ private static class CountingFailingNotifyGateway extends
TestTabletServerGateway {
+ private final AtomicInteger notifyLeaderAndIsrCount = new
AtomicInteger();
+
+ CountingFailingNotifyGateway() {
+ super(true, Collections.emptySet());
+ }
+
+ int getNotifyLeaderAndIsrCount() {
+ return notifyLeaderAndIsrCount.get();
+ }
+
+ @Override
+ public CompletableFuture<NotifyLeaderAndIsrResponse>
notifyLeaderAndIsr(
+ NotifyLeaderAndIsrRequest request) {
+ notifyLeaderAndIsrCount.incrementAndGet();
+ return super.notifyLeaderAndIsr(request);
+ }
+ }
+
/**
* A gateway that intercepts NotifyLeaderAndIsr calls for verifying
sequential execution of
* leader migrations. In pass-through mode, it delegates to the parent. In
controlled mode, it
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
index 0360b6a4e..dfe335fbe 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
@@ -46,6 +46,8 @@ import org.apache.fluss.testutils.common.AllCallbackWrapper;
import org.apache.fluss.utils.clock.ManualClock;
import org.apache.fluss.utils.clock.SystemClock;
import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;
+import org.apache.fluss.utils.concurrent.FlussScheduler;
+import org.apache.fluss.utils.concurrent.Scheduler;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
@@ -84,6 +86,7 @@ public class RebalanceManagerTest {
private LakeTableTieringManager lakeTableTieringManager;
private RebalanceManager rebalanceManager;
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
+ private Scheduler scheduler;
@BeforeAll
static void baseBeforeAll() throws Exception {
@@ -111,6 +114,9 @@ public class RebalanceManagerTest {
TestingMetricGroups.COORDINATOR_METRICS);
kvSnapshotLeaseManager.start();
+ scheduler = new FlussScheduler(1);
+ scheduler.startup();
+
autoPartitionManager =
new AutoPartitionManager(
serverMetadataCache,
@@ -133,6 +139,9 @@ public class RebalanceManagerTest {
@AfterEach
void afterEach() throws Exception {
rebalanceManager.close();
+ if (scheduler != null) {
+ scheduler.shutdown();
+ }
zookeeperClient.deleteRebalanceTask();
metadataManager =
new MetadataManager(
@@ -312,6 +321,7 @@ public class RebalanceManagerTest {
Executors.newFixedThreadPool(1, new
ExecutorThreadFactory("test-coordinator-io")),
metadataManager,
kvSnapshotLeaseManager,
+ scheduler,
SystemClock.getInstance());
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
index 2d3177b2a..630128546 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
@@ -51,7 +51,10 @@ import
org.apache.fluss.shaded.guava32.com.google.common.collect.Sets;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
import org.apache.fluss.utils.clock.SystemClock;
import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;
+import org.apache.fluss.utils.concurrent.FlussScheduler;
+import org.apache.fluss.utils.concurrent.Scheduler;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -94,6 +97,7 @@ class TableBucketStateMachineTest {
private LakeTableTieringManager lakeTableTieringManager;
private CoordinatorMetadataCache serverMetadataCache;
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
+ private Scheduler scheduler;
@BeforeAll
static void baseBeforeAll() throws Exception {
@@ -140,6 +144,16 @@ class TableBucketStateMachineTest {
SystemClock.getInstance(),
TestingMetricGroups.COORDINATOR_METRICS);
kvSnapshotLeaseManager.start();
+
+ scheduler = new FlussScheduler(1);
+ scheduler.startup();
+ }
+
+ @AfterEach
+ void afterEach() throws Exception {
+ if (scheduler != null) {
+ scheduler.shutdown();
+ }
}
@Test
@@ -313,6 +327,7 @@ class TableBucketStateMachineTest {
new Configuration(),
new LakeCatalogDynamicLoader(new
Configuration(), null, true)),
kvSnapshotLeaseManager,
+ scheduler,
SystemClock.getInstance());
CoordinatorEventManager eventManager =
new CoordinatorEventManager(
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 ff6b0246e..6dd9ce7fe 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
@@ -555,6 +555,64 @@ class ReplicaManagerTest extends ReplicaTestBase {
replicaManager.getDiskUsageMonitor().update(0.10);
}
+ @Test
+ void testNewKvLeaderRejectedWhenDiskLocked() throws Exception {
+ TableBucket kvTb = new TableBucket(DATA1_TABLE_ID_PK, 1);
+ TableBucket logTb = new TableBucket(DATA1_TABLE_ID, 1);
+
+ replicaManager.getDiskUsageMonitor().update(0.99);
+ assertThat(replicaManager.isDiskWriteLocked()).isTrue();
+
+ CompletableFuture<List<NotifyLeaderAndIsrResultForBucket>> future =
+ new CompletableFuture<>();
+ replicaManager.becomeLeaderOrFollower(
+ INITIAL_COORDINATOR_EPOCH,
+ Collections.singletonList(
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH_PK),
+ kvTb,
+ Collections.singletonList(TABLET_SERVER_ID),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+
Collections.singletonList(TABLET_SERVER_ID),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH))),
+ future::complete);
+
+ List<NotifyLeaderAndIsrResultForBucket> results = future.get();
+ assertThat(results).hasSize(1);
+ NotifyLeaderAndIsrResultForBucket result = results.get(0);
+
assertThat(result.getError().error()).isEqualTo(Errors.DISK_WRITE_LOCKED);
+ assertThat(result.getError().messageWithFallback()).contains("data
disk usage");
+ Replica kvReplica = replicaManager.getReplicaOrException(kvTb);
+ assertThat(kvReplica.isLeader()).isFalse();
+ assertThat(kvReplica.getKvTablet()).isNull();
+
+ future = new CompletableFuture<>();
+ replicaManager.becomeLeaderOrFollower(
+ INITIAL_COORDINATOR_EPOCH,
+ Collections.singletonList(
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ logTb,
+ Collections.singletonList(TABLET_SERVER_ID),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+
Collections.singletonList(TABLET_SERVER_ID),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH))),
+ future::complete);
+
+ assertThat(future.get()).containsOnly(new
NotifyLeaderAndIsrResultForBucket(logTb));
+
assertThat(replicaManager.getReplicaOrException(logTb).isLeader()).isTrue();
+
+ replicaManager.getDiskUsageMonitor().update(0.10);
+ }
+
@Test
void testPutKv() throws Exception {
TableBucket tb = new TableBucket(DATA1_TABLE_ID_PK, 1);
diff --git a/website/docs/maintenance/configuration.md
b/website/docs/maintenance/configuration.md
index 605bada74..011ae2440 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -56,13 +56,14 @@ during the Fluss cluster working.
## CoordinatorServer
-| Option | Type | Default
| Description
[...]
-|----------------------------------------------------|------------|-----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| coordinator.io-pool.size | Integer | 10
| **Deprecated**: This option is deprecated. Please use `server.io-pool.size`
instead. The size of the IO thread pool to run blocking operations for
coordinator server. This includes discard unnecessary snapshot files. Increase
this value if you experience slow unnecessary snapshot files clean. The default
value is 10.
[...]
-| coordinator.producer-offsets.ttl | Duration | 24h
| The TTL (time-to-live) for producer offsets. Producer offsets older than this
TTL will be automatically cleaned up by the coordinator server. Producer
offsets are used for undo recovery when a Flink job fails over before
completing its first checkpoint. The default value is 24 hours.
[...]
-| coordinator.producer-offsets.cleanup-interval | Duration | 1h
| The interval for cleaning up expired producer offsets and orphan files in
remote storage. The cleanup task runs periodically to remove expired offsets
and any orphan files that may have been left behind due to incomplete
operations. The default value is 1 hour.
[...]
-| coordinator.lifecycle-throttler.inflight-timeout | Duration | 3min
| The timeout for an in-flight drop event in the coordinator's
TableLifecycleThrottler. If a drop event has been admitted but the
corresponding completion callback has not arrived within this timeout, the
throttler abandons tracking of that drop and continues admitting the next
pending drop.
[...]
+| Option | Type |
Default | Description
[...]
+|--------------------------------------------------------|------------|-----------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
+| coordinator.io-pool.size | Integer | 10
| **Deprecated**: This option is deprecated. Please use
`server.io-pool.size` instead. The size of the IO thread pool to run blocking
operations for coordinator server. This includes discard unnecessary snapshot
files. Increase this value if you experience slow unnecessary snapshot files
clean. The default value is 10.
[...]
+| coordinator.producer-offsets.ttl | Duration | 24h
| The TTL (time-to-live) for producer offsets. Producer offsets older than
this TTL will be automatically cleaned up by the coordinator server. Producer
offsets are used for undo recovery when a Flink job fails over before
completing its first checkpoint. The default value is 24 hours.
[...]
+| coordinator.producer-offsets.cleanup-interval | Duration | 1h
| The interval for cleaning up expired producer offsets and orphan files in
remote storage. The cleanup task runs periodically to remove expired offsets
and any orphan files that may have been left behind due to incomplete
operations. The default value is 1 hour.
[...]
+| coordinator.lifecycle-throttler.inflight-timeout | Duration | 3min
| The timeout for an in-flight drop event in the coordinator's
TableLifecycleThrottler. If a drop event has been admitted but the
corresponding completion callback has not arrived within this timeout, the
throttler abandons tracking of that drop and continues admitting the next
pending drop.
[...]
| coordinator.lifecycle-throttler.timeout-check-interval | Duration | 1min
| The periodic interval at which the coordinator's TableLifecycleThrottler
scans in-flight drops for timeouts.
[...]
+| coordinator.offline-leader.retry-delay | Duration | 1min
| The delay before the coordinator retries offline leaders on live tablet
servers after they are marked offline. This lets a leader that was rejected
because of temporary tablet-server conditions, such as disk write protection,
become electable again after recovery.
[...]
## TabletServer
