You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fxbing (via GitHub)" <gi...@apache.org> on 2026/08/10 03:20:41 UTC
[PR] [server] Balance replicas and leaders within each table [fluss]
fxbing opened a new pull request, #3917:
URL: https://github.com/apache/fluss/pull/3917
### Purpose
Linked issue: close #3871
The existing `REPLICA_DISTRIBUTION` and `LEADER_DISTRIBUTION` goals balance replica and leader counts across the whole cluster. As a result, a hot table can still be concentrated on a few TabletServers even when the cluster-wide counts are balanced, causing CPU and request-load hotspots.
This change adds complementary table-level goals so that replica placement and leadership are balanced independently for every table while preserving rack awareness and previously optimized goals.
### Brief change log
- Add `TABLE_REPLICA_DISTRIBUTION` and `TABLE_LEADER_DISTRIBUTION` as additive rebalance goal types.
- Track replicas and leaders by table in `ServerModel`, and expose per-table replica and leader distribution statistics in `ClusterModelStats`.
- Balance each table within its own lower and upper limits. Prefer leadership transfer between existing replicas, and move leader replicas only when leadership transfer cannot reach the table-level target.
- Reject a result when the replica or leader distribution of any individual table regresses, instead of allowing improvements in one table to hide regressions in another.
- Keep the existing cluster-level goals as complementary goals. The recommended order is `RACK_AWARE,TABLE_REPLICA_DISTRIBUTION,REPLICA_DISTRIBUTION,TABLE_LEADER_DISTRIBUTION,LEADER_DISTRIBUTION`.
- Update the Flink procedure and rebalance documentation.
### Tests
- Added unit tests for table replica and leader distribution, partitioned tables, offline servers, balance boundaries, goal ordering, convergence, idempotence, and stable numeric goal mappings.
- Added an end-to-end rebalance scenario covering the recommended five-goal chain.
- Ran 53 related tests on JDK 11: 53 passed, 0 failed.
- Ran Spotless and Checkstyle for `fluss-common`, `fluss-server`, `fluss-client`, and `fluss-flink-common`.
- `RebalanceITCase` compiles locally. Its execution is deferred to CI because the local sandbox does not permit the embedded ZooKeeper server to bind a socket.
### API and Format
- Adds the public goal types `TABLE_REPLICA_DISTRIBUTION(3)` and `TABLE_LEADER_DISTRIBUTION(4)`.
- No RPC schema change: rebalance goals continue to use the existing `repeated int32 goals` field.
- No storage format change.
### Documentation
- Document the new table-level goals and the recommended goal order in the Flink procedure and rebalance operation documentation.
### Generative AI disclosure
- [x] Yes — OpenAI GPT and Anthropic Claude were used for implementation assistance and cross-review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Balance replicas and leaders within each table [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong merged PR #3917:
URL: https://github.com/apache/fluss/pull/3917
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Balance replicas and leaders within each table [fluss]
Posted by "fxbing (via GitHub)" <gi...@apache.org>.
fxbing commented on code in PR #3917:
URL: https://github.com/apache/fluss/pull/3917#discussion_r4236919276
##########
fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/GoalType.java:
##########
@@ -44,7 +44,13 @@ public enum GoalType {
* Goal to generate replica movement tasks to ensure that the number of replicas on each
* tabletServer is near balanced and the replicas are distributed across racks.
*/
- RACK_AWARE(2);
+ RACK_AWARE(2),
+
+ /** Goal to balance replica counts independently for each table. */
+ TABLE_REPLICA_DISTRIBUTION(3),
+
+ /** Goal to balance leader replica counts independently for each table. */
+ TABLE_LEADER_DISTRIBUTION(4);
Review Comment:
@loserwang1024 Added the two-rack test and clarified the guarantees. The combination does not guarantee one leader per server or four per rack: the current window is \[0, 2\], and rack awareness constrains replica placement only. The test also confirms the existing cluster-level goals behave the same way. Stronger balancing can be addressed separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Balance replicas and leaders within each table [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3917:
URL: https://github.com/apache/fluss/pull/3917#discussion_r4236278592
##########
fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/GoalType.java:
##########
@@ -44,7 +44,13 @@ public enum GoalType {
* Goal to generate replica movement tasks to ensure that the number of replicas on each
* tabletServer is near balanced and the replicas are distributed across racks.
*/
- RACK_AWARE(2);
+ RACK_AWARE(2),
+
+ /** Goal to balance replica counts independently for each table. */
+ TABLE_REPLICA_DISTRIBUTION(3),
+
+ /** Goal to balance leader replica counts independently for each table. */
+ TABLE_LEADER_DISTRIBUTION(4);
Review Comment:
Could we clarify how `RACK_AWARE`, `TABLE_REPLICA_DISTRIBUTION`, and `TABLE_LEADER_DISTRIBUTION` work together?
For example, suppose a table has 8 leaders and the cluster has two racks, each containing 4 servers. Ideally, we would have one leader per server and four leaders per rack. Can combining these goals guarantee this distribution? The current per-server bounds appear to allow 0–2 leaders in this case, so an uneven distribution could already be considered balanced.
Could we add a test covering this scenario and clarify what balance guarantees the goal combination provides?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Balance replicas and leaders within each table [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3917:
URL: https://github.com/apache/fluss/pull/3917#issuecomment-6117604264
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Balance replicas and leaders within each table [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3917:
URL: https://github.com/apache/fluss/pull/3917#discussion_r4244615812
##########
fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/GoalType.java:
##########
@@ -44,7 +44,13 @@ public enum GoalType {
* Goal to generate replica movement tasks to ensure that the number of replicas on each
* tabletServer is near balanced and the replicas are distributed across racks.
*/
- RACK_AWARE(2);
+ RACK_AWARE(2),
+
+ /** Goal to balance replica counts independently for each table. */
+ TABLE_REPLICA_DISTRIBUTION(3),
+
+ /** Goal to balance leader replica counts independently for each table. */
+ TABLE_LEADER_DISTRIBUTION(4);
Review Comment:
Ok , I have no problem any more
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Balance replicas and leaders within each table [fluss]
Posted by "fxbing (via GitHub)" <gi...@apache.org>.
fxbing commented on PR #3917:
URL: https://github.com/apache/fluss/pull/3917#issuecomment-5248334452
@swuferhong PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Balance replicas and leaders within each table [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #3917:
URL: https://github.com/apache/fluss/pull/3917#discussion_r4132778734
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/goal/TableLeaderReplicaDistributionGoal.java:
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.rebalance.goal;
+
+import org.apache.fluss.exception.RebalanceFailureException;
+import org.apache.fluss.server.coordinator.rebalance.ActionAcceptance;
+import org.apache.fluss.server.coordinator.rebalance.ActionType;
+import org.apache.fluss.server.coordinator.rebalance.RebalancingAction;
+import org.apache.fluss.server.coordinator.rebalance.model.BucketModel;
+import org.apache.fluss.server.coordinator.rebalance.model.ClusterModel;
+import org.apache.fluss.server.coordinator.rebalance.model.ClusterModelStats;
+import org.apache.fluss.server.coordinator.rebalance.model.ReplicaModel;
+import org.apache.fluss.server.coordinator.rebalance.model.ServerModel;
+import org.apache.fluss.utils.MathUtils;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import static org.apache.fluss.server.coordinator.rebalance.ActionAcceptance.ACCEPT;
+import static org.apache.fluss.server.coordinator.rebalance.ActionAcceptance.REPLICA_REJECT;
+import static org.apache.fluss.utils.MathUtils.EPSILON;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Soft goal to distribute leader replicas of every table across tablet servers. */
+public class TableLeaderReplicaDistributionGoal extends TableDistributionAbstractGoal {
+
+ @Override
+ public ActionAcceptance actionAcceptance(RebalancingAction action, ClusterModel clusterModel) {
+ ServerModel sourceServer = clusterModel.server(action.getSourceServerId());
+ ServerModel destinationServer = clusterModel.server(action.getDestinationServerId());
+ checkNotNull(
+ sourceServer, "Source server " + action.getSourceServerId() + " is not found.");
+ checkNotNull(
+ destinationServer,
+ "Destination server " + action.getDestinationServerId() + " is not found.");
+ ReplicaModel sourceReplica = sourceServer.replica(action.getTableBucket());
+ checkNotNull(sourceReplica, "Source replica " + action.getTableBucket() + " is not found.");
+ if (action.getActionType() == ActionType.REPLICA_MOVEMENT && !sourceReplica.isLeader()) {
+ return ACCEPT;
+ }
+ if (action.getActionType() != ActionType.LEADERSHIP_MOVEMENT
+ && action.getActionType() != ActionType.REPLICA_MOVEMENT) {
+ throw new IllegalArgumentException("Unsupported action type " + action.getActionType());
+ }
+ long tableId = action.getTableBucket().getTableId();
+ return destinationServer.numLeaderReplicas(tableId) + 1
+ <= upperLimit(tableId, destinationServer)
+ && (!isAlive(sourceServer)
+ || sourceServer.numLeaderReplicas(tableId) - 1
+ >= lowerLimit(tableId, sourceServer))
+ ? ACCEPT
+ : REPLICA_REJECT;
+ }
+
+ @Override
+ protected void rebalanceForServer(
+ ServerModel server, ClusterModel clusterModel, Set<Goal> optimizedGoals)
+ throws RebalanceFailureException {
+ for (Long tableId : clusterModel.tables()) {
+ int numLeaders = server.numLeaderReplicas(tableId);
+ if (numLeaders > upperLimit(tableId, server) || server.isOfflineTagged()) {
+ boolean leadershipNotMoved =
+ rebalanceByMovingLeadershipOut(
+ server, tableId, clusterModel, optimizedGoals);
+ boolean leadersNotMoved =
+ leadershipNotMoved
+ && rebalanceByMovingLeaderReplicasOut(
+ server, tableId, clusterModel, optimizedGoals);
Review Comment:
Could we defer replica movement until leadership-only balancing has stopped making progress across all servers of the table? Failing to transfer more leaders from the current server does not mean that data movement is necessary: another server may first need to transfer a leader away to create room.
For example, consider one table with 12 buckets on four servers, with the first server in each replica list being the leader:
- 6 buckets: `[S0, S1]`
- 3 buckets: `[S1, S2]`
- 1 bucket: `[S2, S3]`
- 2 buckets: `[S3, S1]`
The initial leader counts are `[6, 3, 1, 2]`, and the balance window is `[2, 4]`. Running this goal alone, S0 first transfers one leader to S1, giving `[5, 4, 1, 2]`. Since S1 is now at the upper limit, the current code immediately moves a leader replica from S0 to S2.
However, transferring an existing leader from S1 to S2 and then another from S0 to S1 would reach `[4, 4, 2, 2]` without moving any data.
Could we introduce a separate leadership-only phase, repeated across all servers until no further progress is possible, before falling back to replica movement? This would avoid an unnecessary data copy in this case.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/goal/TableReplicaDistributionGoal.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * 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.rebalance.goal;
+
+import org.apache.fluss.exception.RebalanceFailureException;
+import org.apache.fluss.server.coordinator.rebalance.ActionAcceptance;
+import org.apache.fluss.server.coordinator.rebalance.ActionType;
+import org.apache.fluss.server.coordinator.rebalance.RebalancingAction;
+import org.apache.fluss.server.coordinator.rebalance.model.ClusterModel;
+import org.apache.fluss.server.coordinator.rebalance.model.ClusterModelStats;
+import org.apache.fluss.server.coordinator.rebalance.model.ReplicaModel;
+import org.apache.fluss.server.coordinator.rebalance.model.ServerModel;
+import org.apache.fluss.utils.MathUtils;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import static org.apache.fluss.server.coordinator.rebalance.ActionAcceptance.ACCEPT;
+import static org.apache.fluss.server.coordinator.rebalance.ActionAcceptance.REPLICA_REJECT;
+import static org.apache.fluss.utils.MathUtils.EPSILON;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Soft goal to distribute replicas of every table across tablet servers. */
+public class TableReplicaDistributionGoal extends TableDistributionAbstractGoal {
+
+ @Override
+ public ActionAcceptance actionAcceptance(RebalancingAction action, ClusterModel clusterModel) {
+ if (action.getActionType() == ActionType.LEADERSHIP_MOVEMENT) {
+ return ACCEPT;
+ }
+ if (action.getActionType() != ActionType.REPLICA_MOVEMENT) {
+ throw new IllegalArgumentException("Unsupported action type " + action.getActionType());
+ }
+ ServerModel sourceServer = clusterModel.server(action.getSourceServerId());
+ ServerModel destinationServer = clusterModel.server(action.getDestinationServerId());
+ checkNotNull(
+ sourceServer, "Source server " + action.getSourceServerId() + " is not found.");
+ checkNotNull(
+ destinationServer,
+ "Destination server " + action.getDestinationServerId() + " is not found.");
+ long tableId = action.getTableBucket().getTableId();
+ return destinationServer.numReplicas(tableId) + 1 <= upperLimit(tableId, destinationServer)
+ && (!isAlive(sourceServer)
+ || sourceServer.numReplicas(tableId) - 1
+ >= lowerLimit(tableId, sourceServer))
+ ? ACCEPT
+ : REPLICA_REJECT;
+ }
+
+ @Override
+ protected void rebalanceForServer(
+ ServerModel server, ClusterModel clusterModel, Set<Goal> optimizedGoals)
+ throws RebalanceFailureException {
+ for (Long tableId : clusterModel.tables()) {
+ int replicaCount = server.numReplicas(tableId);
+ if (replicaCount > upperLimit(tableId, server) || server.isOfflineTagged()) {
+ if (rebalanceByMovingReplicasOut(server, tableId, clusterModel, optimizedGoals)) {
+ tablesAboveRebalanceUpperLimit.add(tableId);
+ }
+ } else if (isAlive(server) && replicaCount < lowerLimit(tableId, server)) {
+ if (rebalanceByMovingReplicasIn(server, tableId, clusterModel, optimizedGoals)) {
+ tablesBelowRebalanceLowerLimit.add(tableId);
+ }
+ }
+ }
+ }
+
+ @Override
+ public ClusterModelStatsComparator clusterModelStatsComparator() {
+ return new TableReplicaDistributionStatsComparator();
+ }
+
+ @Override
+ Map<Long, Integer> numInterestedReplicasByTable(ClusterModel clusterModel) {
+ return clusterModel.numReplicasByTable();
+ }
+
+ private boolean rebalanceByMovingReplicasOut(
+ ServerModel sourceServer,
+ long tableId,
+ ClusterModel clusterModel,
+ Set<Goal> optimizedGoals) {
+ SortedSet<ServerModel> candidates =
+ new TreeSet<>(
+ Comparator.comparingInt((ServerModel server) -> server.numReplicas(tableId))
+ .thenComparingInt(ServerModel::numReplicas)
+ .thenComparingInt(ServerModel::id));
+ for (ServerModel server : clusterModel.aliveServers()) {
+ if (server.numReplicas(tableId) < upperLimit(tableId, server)) {
+ candidates.add(server);
+ }
+ }
+ for (ReplicaModel replica : sourceServer.replicas(tableId)) {
+ ServerModel destination =
+ maybeApplyBalancingAction(
+ clusterModel,
+ replica,
+ candidates,
+ ActionType.REPLICA_MOVEMENT,
+ optimizedGoals);
+ if (destination != null) {
+ if (sourceServer.numReplicas(tableId) <= upperLimit(tableId, sourceServer)) {
+ return false;
+ }
+ candidates.remove(destination);
+ if (destination.numReplicas(tableId) < upperLimit(tableId, destination)) {
+ candidates.add(destination);
+ }
Review Comment:
`maybeApplyBalancingAction()` updates the destination's replica count before this `remove()` call, but the `TreeSet` comparator uses that count as its sort key. Mutating the key while the element is still in the tree can make `remove(destination)` fail. Adding it again can then insert the same server a second time and leave the candidates incorrectly ordered.
For example, starting with four candidates S1–S4 at zero replicas, moving a replica to S1 changes its sort key from 0 to 1. I reproduced a case where removal failed and reinsertion produced five entries, with S1 appearing twice. In a rebalance with 20 replicas initially on S0 and four empty destinations, the first five moves all went to S1 while the other destinations remained empty.
Could we remove the candidate before mutating its count, use immutable ordering entries, or rebuild the ordering after each successful move? The same issue exists in `TableLeaderReplicaDistributionGoal.rebalanceByMovingLeaderReplicasOut()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org