[server] Support configurable concurrent rebalance tasks

fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceStatus.java

@@ -38,7 +38,7 @@ public enum RebalanceStatus {
     TIMEOUT(5);
 
     public static final Set<RebalanceStatus> FINAL_STATUSES =
-            new HashSet<>(Arrays.asList(COMPLETED, CANCELED, FAILED, TIMEOUT));
+            new HashSet<>(Arrays.asList(COMPLETED, CANCELED, FAILED));
 
     private final int code;
 

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

@@ -251,6 +251,19 @@ public class ConfigOptions {
                                     + "conditions, such as disk write protection, become electable "
                                     + "again after recovery.");
 
+    public static final ConfigOption<Integer> COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS =
+            key("coordinator.rebalance.max-inflight-tasks")
+                    .intType()
+                    .defaultValue(1)
+                    .withDescription(
+                            "The maximum number of bucket-level rebalance tasks that can be "
+                                    + "executed concurrently by the coordinator. A higher value "
+                                    + "can speed up rebalance, while a lower value reduces the "
+                                    + "number of simultaneous bucket movements. Setting it to 0 "
+                                    + "pauses scheduling new rebalance tasks; already in-flight "
+                                    + "tasks continue until they complete or time out. The value "
+                                    + "must be non-negative.");
+
     public static final ConfigOption<Boolean> LOG_TABLE_ALLOW_CREATION =
             key("allow.create.log.tables")
                     .booleanType()

fluss-common/src/test/java/org/apache/fluss/cluster/rebalance/RebalanceStatusTest.java

@@ -0,0 +1,37 @@
+/*
+ * 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.cluster.rebalance;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.CANCELED;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.COMPLETED;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FAILED;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FINAL_STATUSES;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.TIMEOUT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link RebalanceStatus}. */
+class RebalanceStatusTest {
+
+    @Test
+    void testTimeoutIsRecoverableRatherThanFinal() {
+        assertThat(FINAL_STATUSES).containsExactlyInAnyOrder(COMPLETED, CANCELED, FAILED);
+        assertThat(FINAL_STATUSES).doesNotContain(TIMEOUT);
+    }
+}

fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java

@@ -99,6 +99,21 @@ public void register(ServerReconfigurable serverReconfigurable) {
         dynamicServerConfig.register(serverReconfigurable);
     }
 
+    /**
+     * Register a ServerReconfigurable and immediately apply the current effective configuration.
+     *
+     * <p>Use this for components created after {@link #startup()}, when persisted dynamic
+     * configuration may already have been loaded.
+     */
+    public void registerAndApplyCurrentConfig(ServerReconfigurable serverReconfigurable) {
+        dynamicServerConfig.registerAndApplyCurrentConfig(serverReconfigurable);
+    }
+
+    /** Unregister a ServerReconfigurable that no longer listens to configuration changes. */
+    public void unregister(ServerReconfigurable serverReconfigurable) {
+        dynamicServerConfig.unregister(serverReconfigurable);
+    }
+
     /**
      * Register a ConfigValidator for stateless validation.
      *

fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java

@@ -45,6 +45,7 @@
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import static org.apache.fluss.config.ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS;
 import static org.apache.fluss.config.ConfigOptions.DATALAKE_FORMAT;
 import static org.apache.fluss.config.ConfigOptions.KV_LEADER_REPLICA_MEMORY_RESERVED;
 import static org.apache.fluss.config.ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC;
@@ -77,6 +78,7 @@ class DynamicServerConfig {
             new HashSet<>(
                     Arrays.asList(
                             DATALAKE_FORMAT.key(),
+                            COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS.key(),
                             LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED.key(),
                             LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(),
                             KV_LEADER_REPLICA_MEMORY_RESERVED.key(),
@@ -137,6 +139,24 @@ void register(ServerReconfigurable serverReconfigurable) {
         serverReconfigures.put(serverReconfigurable.getClass(), serverReconfigurable);
     }
 
+    void registerAndApplyCurrentConfig(ServerReconfigurable serverReconfigurable) {
+        inWriteLock(
+                lock,
+                () -> {
+                    serverReconfigurable.validate(currentConfig);
+                    serverReconfigurable.reconfigure(currentConfig);
+                    serverReconfigures.put(serverReconfigurable.getClass(), serverReconfigurable);
+                });
+    }
+
+    void unregister(ServerReconfigurable serverReconfigurable) {
+        inWriteLock(
+                lock,
+                () ->
+                        serverReconfigures.remove(
+                                serverReconfigurable.getClass(), serverReconfigurable));
+    }
+
     /**
      * Register a ConfigValidator for stateless validation.
      *