imbajin commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4295044334
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/config/PDConfig.java:
##########
@@ -58,6 +58,16 @@ public class PDConfig implements InitializingBean {
// The patrol task interval
@Value("${pd.patrol-interval:300}")
private long patrolInterval = 300;
+
+ // Periodic partition leader balancing, off by default; the interval is in
seconds
+ @Value("${pd.balance-leader-enable:false}")
+ private boolean balanceLeaderEnable = false;
+ @Value("${pd.balance-leader-interval:300}")
+ private long balanceLeaderInterval = 300;
+ // Skip the periodic run while max - min leaders per store is at or below
this
+ @Value("${pd.balance-leader-max-skew:2}")
+ private int balanceLeaderMaxSkew = 2;
Review Comment:
⚠️ Important: This setting accepts negative values without validation. With
`pd.balance-leader-max-skew=-1`, even a balanced cluster has `skew=0`, so every
enabled timer tick bypasses the early return and invokes the planner, which can
transfer leaders to placement-derived targets. Please reject negative values
during config initialization and document the allowed range.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +607,80 @@ public synchronized Map<Integer, Long>
balancePartitionLeader(boolean immediatel
return results;
}
+ /**
+ * The periodic entry of leader balancing. Calls balancePartitionLeader
only when the
+ * leader count per store has drifted further apart than
pd.balance-leader-max-skew, so a
+ * balanced cluster sees no transfers. Skipped while a shard group is not
Normal, has no
+ * recent partition heartbeat or has a shard that is neither Leader nor
Follower, or while a
+ * store that holds a shard is not active, because the target distribution
would include a
+ * replica that cannot take a leader.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty
when skipped
+ */
+ public Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ if (!isLeader()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ Set<Long> activeStores = storeService.getActiveStores().stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ long now = System.currentTimeMillis();
+ Map<Long, Integer> leaderCount = new HashMap<>();
+ for (Metapb.ShardGroup group : shardGroups) {
+ // A store stays active for keepAlive-timeout after it stops, but
partition
+ // heartbeats from the surviving replicas already mark its groups
Warn.
+ // balancePartitionLeader plans over all groups, so skip the whole
run.
+ if (group.getState() != Metapb.PartitionState.PState_Normal) {
+ log.info("periodic leader balance skipped, shard group {} is
{}",
+ group.getId(), group.getState());
+ return new HashMap<>();
+ }
+ // A group that lost its quorum stops sending stats and keeps its
last state, so
+ // the state above only counts while the group's leader is still
reporting it
+ Metapb.PartitionStats stats =
partitionService.getPartitionStats("", group.getId());
+ if (stats == null || now - stats.getTimestamp() >
BALANCE_LEADER_STATS_MAX_AGE) {
+ log.info("periodic leader balance skipped, shard group {} has
no partition " +
+ "heartbeat in the last {} ms", group.getId(),
+ BALANCE_LEADER_STATS_MAX_AGE);
+ return new HashMap<>();
+ }
+ for (Metapb.Shard shard : group.getShardsList()) {
+ // reallocShards adds a shard with role None while the
membership change runs;
+ // balancePartitionLeader would pick it, or a Learner, as a
transfer target
+ if (shard.getRole() != Metapb.ShardRole.Leader &&
Review Comment:
⚠️ Important: `Follower` confirms a voting role, but not that the replica
has applied the leader's latest log. The fresh group heartbeat and active-Store
checks do not expose that lag. With JRaft 1.3.14, accepting a transfer moves
the leader to `STATE_TRANSFERRING` while it waits for the target to catch up,
and writes are rejected in that state; an auto-balance that picks a live but
lagging follower can therefore pause writes until catch-up or transfer timeout.
Please defer transfers until the target's replication progress is caught up and
cover a live-but-behind follower.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +607,80 @@ public synchronized Map<Integer, Long>
balancePartitionLeader(boolean immediatel
return results;
}
+ /**
+ * The periodic entry of leader balancing. Calls balancePartitionLeader
only when the
+ * leader count per store has drifted further apart than
pd.balance-leader-max-skew, so a
+ * balanced cluster sees no transfers. Skipped while a shard group is not
Normal, has no
+ * recent partition heartbeat or has a shard that is neither Leader nor
Follower, or while a
+ * store that holds a shard is not active, because the target distribution
would include a
+ * replica that cannot take a leader.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty
when skipped
+ */
+ public Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ if (!isLeader()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ Set<Long> activeStores = storeService.getActiveStores().stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ long now = System.currentTimeMillis();
+ Map<Long, Integer> leaderCount = new HashMap<>();
+ for (Metapb.ShardGroup group : shardGroups) {
+ // A store stays active for keepAlive-timeout after it stops, but
partition
+ // heartbeats from the surviving replicas already mark its groups
Warn.
+ // balancePartitionLeader plans over all groups, so skip the whole
run.
+ if (group.getState() != Metapb.PartitionState.PState_Normal) {
+ log.info("periodic leader balance skipped, shard group {} is
{}",
+ group.getId(), group.getState());
+ return new HashMap<>();
+ }
+ // A group that lost its quorum stops sending stats and keeps its
last state, so
+ // the state above only counts while the group's leader is still
reporting it
+ Metapb.PartitionStats stats =
partitionService.getPartitionStats("", group.getId());
+ if (stats == null || now - stats.getTimestamp() >
BALANCE_LEADER_STATS_MAX_AGE) {
+ log.info("periodic leader balance skipped, shard group {} has
no partition " +
+ "heartbeat in the last {} ms", group.getId(),
+ BALANCE_LEADER_STATS_MAX_AGE);
+ return new HashMap<>();
+ }
+ for (Metapb.Shard shard : group.getShardsList()) {
+ // reallocShards adds a shard with role None while the
membership change runs;
+ // balancePartitionLeader would pick it, or a Learner, as a
transfer target
+ if (shard.getRole() != Metapb.ShardRole.Leader &&
+ shard.getRole() != Metapb.ShardRole.Follower) {
+ log.info("periodic leader balance skipped, shard group {}
has store {} " +
+ "as {}", group.getId(), shard.getStoreId(),
shard.getRole());
+ return new HashMap<>();
+ }
+ if (!activeStores.contains(shard.getStoreId())) {
+ log.info("periodic leader balance skipped, store {} of
shard group {} is " +
+ "not active", shard.getStoreId(), group.getId());
+ return new HashMap<>();
+ }
+ int leader = shard.getRole() == Metapb.ShardRole.Leader ? 1 :
0;
+ leaderCount.merge(shard.getStoreId(), leader, Integer::sum);
+ }
+ }
+
+ int skew = Collections.max(leaderCount.values()) -
Collections.min(leaderCount.values());
+ if (skew <= pdConfig.getBalanceLeaderMaxSkew()) {
+ log.debug("periodic leader balance skipped, leader count by store:
{}, skew {}",
+ leaderCount, skew);
+ return new HashMap<>();
+ }
+
+ log.info("leader count by store: {}, skew {} exceeds {}, balancing
leaders",
+ leaderCount, skew, pdConfig.getBalanceLeaderMaxSkew());
+ return balancePartitionLeader(false);
Review Comment:
⚠️ Important: The preflight validates one `getShardGroups()` snapshot, but
this call enters `balancePartitionLeader(false)`, which reloads groups and
selects any non-Leader shard as a target. If `reallocShards()` adds a `None`
shard between those reads and starts its membership adjustment, the planner can
send leadership to that not-yet-voting peer despite this guard. Please plan
from the validated snapshot or recheck target membership and role immediately
before dispatch.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
