You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hugegraph.apache.org by "arshilkxwork (via GitHub)" <gi...@apache.org> on 2026/10/15 12:53:31 UTC
[PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
arshilkxwork opened a new pull request, #3356:
URL: https://github.com/apache/hugegraph/pull/3356
## Purpose of the PR
- close #3135
PD balances partition leaders only when `/v1/task/balanceLeaders`, `/v1/balanceLeaders` or the `balanceLeaders` gRPC call is made. When a Store restarts, raft elects new leaders for its groups on the surviving Stores, and nothing moves them back. On Kubernetes, where pod restarts are routine, leader-side work stays concentrated on the Stores that did not restart until someone calls the endpoint by hand.
## Main Changes
- `PDConfig`: three new keys, `pd.balance-leader-enable` (default `false`), `pd.balance-leader-interval` (seconds, default `300`) and `pd.balance-leader-max-skew` (default `2`). Names follow the proposal in the issue. With the defaults, behavior does not change.
- `TaskScheduleService.init()` schedules `autoBalancePartitionLeader()` with `scheduleWithFixedDelay`, like the other periodic jobs, when the feature is enabled. A non-positive interval logs a warning and schedules nothing, so a bad value cannot fail PD startup with the executor's `IllegalArgumentException`.
- `autoBalancePartitionLeader()` does nothing on a follower PD. On the leader it counts leaders per Store from the shard groups. A Store that holds shards but leads none counts as 0. If max minus min is at or below `max-skew` it returns without touching anything. Otherwise it calls the existing `balancePartitionLeader(false)`, so the algorithm, the 30 s rate limit and the refusal during split, move and shard-balance tasks are the same as for a manual call. The scheduled wrapper logs those refusals (`PDException` code 1001) at INFO level.
- One guard the manual path does not have: a run is skipped while any Store that holds a shard is missing from the active-store list. `balancePartitionLeader` computes its target from shard placement, so with a Store down it would try to transfer leaders to that Store. The periodic job waits until every Store holding a replica is heartbeating again.
- Docs: the new keys in `hugegraph-pd/docs/configuration.md` and `architecture.md`, commented defaults in the shipped `hg-pd-dist` `application.yml`, and a note in the Store operations guide next to the manual `balanceLeaders` call.
Not included: the event-driven trigger on a Store returning to `Up`, which the issue lists as a possible follow-up. The skew check uses max minus min over the Stores that hold shards, as proposed in the issue. With uneven shard placement, the best reachable distribution can itself exceed `max-skew`. In that case each run calls `balancePartitionLeader`, which finds every leader already on its target and transfers nothing.
## Verifying these changes
- [ ] Trivial rework / code cleanup without any test coverage. (No Need)
- [ ] Already covered by existing tests, such as *(please modify tests here)*.
- [x] Need tests and can be verified as follows:
- `TaskScheduleServiceTest` (core) gets three cases on six shard groups over Stores 1-3: leaders spread 2/2/2 makes no transfers; all six leaders on Store 1 moves four leaders to Stores 2 and 3 and none to Store 1; Store 3 without a heartbeat skips the run.
- `mvn test -pl hugegraph-pd/hg-pd-test -P pd-core-test`: 153 tests, 0 failures, 2 skipped (both skips predate this change).
- Not tested against a running PD and Store cluster.
## Does this PR potentially affect the following parts?
- [ ] Dependencies ([add/update license](https://hugegraph.apache.org/docs/contribution-guidelines/contribute/#321-check-licenses) info & [regenerate_known_dependencies.sh](../install-dist/scripts/dependency/regenerate_known_dependencies.sh))
- [x] Modify configurations
- [ ] The public API
- [ ] Other affects (typed here)
- [ ] Nope
## Documentation Status
- [ ] `Doc - TODO`: required documentation is pending; complete it before merging.
- [x] `Doc - Done`: documentation is included here or linked below.
- [ ] `Doc - No Need`: no user-visible documentation is affected.
Documentation files in this PR or paired hugegraph-doc PR:
- `hugegraph-pd/docs/configuration.md`
- `hugegraph-pd/docs/architecture.md`
- `hugegraph-pd/hg-pd-dist/src/assembly/static/conf/application.yml`
- `hugegraph-store/docs/operations-guide.md`
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
bitflicker64 commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4304748201
##########
hugegraph-pd/docs/configuration.md:
##########
@@ -196,6 +202,21 @@ pd:
- Production: `1800` (30 minutes) to reduce overhead
- Large clusters: `3600` (1 hour)
- `pd.initial-store-count`: Set to expected initial store count (e.g., `3` for 3 stores)
+- `pd.balance-leader-enable`: After a Store restarts, its raft groups keep the leaders they elected
+ on the surviving Stores. Enable this so PD moves leaders back without a manual
+ `balanceLeaders` call. A check is skipped while any shard group is not in the `Normal` state,
+ has sent no partition heartbeat in the last 30 seconds, or has a replica that is not a Leader
+ or Follower (a membership change in progress); while a Store that holds a shard is not sending
+ heartbeats; and while a split, move or shard balance task is running. A planned transfer is
Review Comment:
Minor: "while a Store that holds a shard is not sending heartbeats" reads as if one missed heartbeat stops the check. The code checks the active-store list, and a Store stays on it for `store.keepAlive-timeout` (300 s by default) after its last heartbeat, as the PR description says.
Please write "while a Store that holds a shard is not in the active-store list, which it leaves `store.keepAlive-timeout` after its last heartbeat".
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +659,172 @@ 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. The plan uses the groups checked here; a group keeps
+ * its leader when this PD is no longer the leader, the group changed, or the target trails
+ * the leader by the time the transfer would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
+ Set<Long> activeStores = stores.stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ // shard group id -> store id -> the applied index from the store's last heartbeat
+ Map<Integer, Map<Long, IndexReport>> indexReports = new HashMap<>();
+ for (Metapb.Store store : stores) {
+ for (Metapb.RaftStats raftStats : store.getStats().getRaftStatsList()) {
+ indexReports.computeIfAbsent(raftStats.getPartitionId(), id -> new HashMap<>())
+ .put(store.getId(), new IndexReport(store.getId(),
+ raftStats.getCommittedIndex(),
+ store.getLastHeartbeat()));
+ }
+ }
+ // Plan with the leader reports of the previous run and keep this run's for the next,
+ // including runs that skip below, so a later run can tell the leader's write rate
+ Map<Integer, IndexReport> previousLeaderReports = new HashMap<>(lastLeaderIndexReports);
+ lastLeaderIndexReports.clear();
+ for (Metapb.ShardGroup group : shardGroups) {
+ Long leaderStore = getLeaderStore(group);
+ IndexReport report = leaderStore == null ? null :
+ indexReports.getOrDefault(group.getId(), Map.of())
+ .get(leaderStore);
+ if (report != null) {
+ lastLeaderIndexReports.put(group.getId(), report);
+ }
+ }
+ 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 <= maxSkew) {
+ 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, maxSkew);
+ return balancePartitionLeader(false, shardGroups, (group, storeId) ->
+ canTakeLeader(group, storeId, indexReports, previousLeaderReports));
+ }
+
+ private static Long getLeaderStore(Metapb.ShardGroup group) {
+ return group.getShardsList().stream()
+ .filter(shard -> shard.getRole() == Metapb.ShardRole.Leader)
+ .map(Metapb.Shard::getStoreId)
+ .findFirst().orElse(null);
+ }
+
+ /**
+ * Checks a periodic transfer right before it is sent: this PD must still be the leader, the
+ * group must be unchanged since autoBalancePartitionLeader checked it, and the target's
+ * applied index must trail the leader's by at most BALANCE_LEADER_MAX_INDEX_GAP. That gap is
+ * estimated from store heartbeats taken at different times, with the leader's progress in
+ * between taken as its average rate since the previous run, so it is not a strict bound:
+ * a burst late in that time makes a caught-up target look behind, and a burst early in it
+ * hides a target that fell behind.
+ */
+ private boolean canTakeLeader(Metapb.ShardGroup group, long storeId,
+ Map<Integer, Map<Long, IndexReport>> indexReports,
+ Map<Integer, IndexReport> previousLeaderReports) throws
+ PDException {
+ // A PD that lost leadership during the run must not send what it planned as leader
+ if (!isLeader()) {
+ log.info("periodic leader balance keeps the leader of shard group {}, this PD is " +
+ "no longer the leader", group.getId());
+ return false;
+ }
+ if (!group.equals(storeService.getShardGroup(group.getId()))) {
+ log.info("periodic leader balance keeps the leader of shard group {}, the group " +
+ "changed after the check", group.getId());
+ return false;
+ }
+ Long leaderStore = getLeaderStore(group);
+ Map<Long, IndexReport> reports = indexReports.getOrDefault(group.getId(), Map.of());
+ IndexReport leader = leaderStore == null ? null : reports.get(leaderStore);
+ IndexReport target = reports.get(storeId);
+ if (leader == null || target == null) {
+ log.info("periodic leader balance keeps the leader of shard group {}, no applied " +
+ "index reported by store {} or leader store {}", group.getId(), storeId,
+ leaderStore);
+ return false;
+ }
+ // Stores send their heartbeats at different times, so a caught-up target reporting
+ // earlier than the leader looks behind by the writes in between. Estimate the leader's
+ // index at the time of the target's report, at the average rate it grew since the
+ // previous run. Without that rate, compare the indexes as reported.
+ long leaderIndex = leader.index();
+ IndexReport previous = previousLeaderReports.get(group.getId());
+ if (previous != null && previous.storeId() == leader.storeId() &&
+ previous.time() < leader.time() && previous.index() <= leader.index()) {
+ double rate = (double) (leader.index() - previous.index()) /
+ (leader.time() - previous.time());
+ leaderIndex += Math.round(rate * (target.time() - leader.time()));
Review Comment:
Important: Nothing limits the age of the target report this line backdates to. The description accepts a follower that fell behind after its last report, which is at most one 30 s heartbeat. A Store stays active for `store.keepAlive-timeout`, 300 s by default. If its heartbeats stop reaching PD while its raft peer is alive but behind (an overloaded Store, say), its groups stay `Normal`, and a minutes-old report passes. Without history, the same report is deferred. The leader then rejects writes until JRaft ends the transfer (`electionTimeoutMs`, 3000), which the 1000-entry check was added to prevent.
Temporary test at b471ea8a: after `warmUp(service, 70_000, 300_000)`, leaders at 100000 now, store 3 at 75000 reported 250 s ago. With history: `{2=2, 3=2, 4=3, 5=3}`. Without: `{2=2, 3=2}`.
Please skip a target whose report is older than about 60 s (two store heartbeats), or cap the backdating at 30 s, and add the case to the lag tests.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#issuecomment-6192830258
Review at f57f45da673a5450f17c1604c44eb8189a842b9d: 7/10 overall (independent reviewers: 7 and 6.5). The opt-in implementation is compact, but please address one P2 availability issue before merging.
P2 — Skip automatic balancing when shard groups are known unhealthy. In TaskScheduleService.java:630–636, active-store membership is checked but group.getState() is ignored. Active entries survive for the configured keepAliveTimeout (300 seconds by default). During a normal Store outage, surviving partition heartbeats can already mark a three-replica group PState_Warn while the missing Store remains in that active set. A scheduled tick can therefore pass this guard and ask the existing global balancer to transfer leaders to that unavailable replica. The timer is not synchronized with Store failures, so its default 300-second interval does not prevent this window.
This can affect writes, rather than merely produce a failed-transfer log: the pinned JRaft 1.3.14 transfer path can enter STATE_TRANSFERRING while waiting for the target and reject writes until the election timeout. The manual transfer behavior predates this PR; the new automatic trigger exposes it without an operator choosing the timing.
A minimal bounded fix is to skip the automatic pass when any shard group is non-Normal, since balancePartitionLeader(false) plans globally. Add a regression with all Store IDs still active but one group Warn: expect zero transfers; after restoring Normal state, the same imbalance should trigger balancing. This guards known-unhealthy metadata without claiming complete replica-readiness detection or requiring an allocator rewrite.
Evidence: source inspection and independent cross-review, not a live-cluster reproduction. The new three tests cover balanced, imbalanced and missing-active-store cases; upstream PD core tests passed during review, while the overall workflow was still running.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4291659870
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +605,61 @@ 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 or a store
+ * that holds a shard is not active, because the target distribution would include a store
+ * 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());
+ 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) {
Review Comment:
Done in fac2cc330. Each group now also needs partition stats received in the last 30 seconds, using the timestamp PD sets in `partitionHeartbeat` (Stores send stats every 5 seconds). A group with no stats at all also skips the run.
`testAutoBalancePartitionLeaderWithoutQuorum` keeps group 0 `Normal` and every Store active, then gives group 0 stats 60 seconds old, and then none: no transfers either way. With stats 10 seconds old the same layout moves four leaders.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +605,61 @@ 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 or a store
+ * that holds a shard is not active, because the target distribution would include a store
+ * 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());
+ 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<>();
+ }
+ for (Metapb.Shard shard : group.getShardsList()) {
+ 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;
Review Comment:
Done in fac2cc330. The run is skipped while any shard in any group has a role other than `Leader` or `Follower`, which covers the `None` shard `reallocShards` adds and a `Learner`. I skip the whole run instead of filtering targets, since `balancePartitionLeader` plans all groups together.
`testAutoBalancePartitionLeaderWithPendingMember` sets store 3's shard in group 0 to `None`, then `Learner`, and expects no transfers. PD core suite: 156 tests, 0 failures, 2 skipped.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4305939970
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +659,172 @@ 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. The plan uses the groups checked here; a group keeps
+ * its leader when this PD is no longer the leader, the group changed, or the target trails
+ * the leader by the time the transfer would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
+ Set<Long> activeStores = stores.stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ // shard group id -> store id -> the applied index from the store's last heartbeat
+ Map<Integer, Map<Long, IndexReport>> indexReports = new HashMap<>();
+ for (Metapb.Store store : stores) {
+ for (Metapb.RaftStats raftStats : store.getStats().getRaftStatsList()) {
+ indexReports.computeIfAbsent(raftStats.getPartitionId(), id -> new HashMap<>())
+ .put(store.getId(), new IndexReport(store.getId(),
+ raftStats.getCommittedIndex(),
+ store.getLastHeartbeat()));
+ }
+ }
+ // Plan with the leader reports of the previous run and keep this run's for the next,
+ // including runs that skip below, so a later run can tell the leader's write rate
+ Map<Integer, IndexReport> previousLeaderReports = new HashMap<>(lastLeaderIndexReports);
+ lastLeaderIndexReports.clear();
+ for (Metapb.ShardGroup group : shardGroups) {
+ Long leaderStore = getLeaderStore(group);
+ IndexReport report = leaderStore == null ? null :
+ indexReports.getOrDefault(group.getId(), Map.of())
+ .get(leaderStore);
+ if (report != null) {
+ lastLeaderIndexReports.put(group.getId(), report);
+ }
+ }
+ 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 <= maxSkew) {
+ 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, maxSkew);
+ return balancePartitionLeader(false, shardGroups, (group, storeId) ->
+ canTakeLeader(group, storeId, indexReports, previousLeaderReports));
+ }
+
+ private static Long getLeaderStore(Metapb.ShardGroup group) {
+ return group.getShardsList().stream()
+ .filter(shard -> shard.getRole() == Metapb.ShardRole.Leader)
+ .map(Metapb.Shard::getStoreId)
+ .findFirst().orElse(null);
+ }
+
+ /**
+ * Checks a periodic transfer right before it is sent: this PD must still be the leader, the
+ * group must be unchanged since autoBalancePartitionLeader checked it, and the target's
+ * applied index must trail the leader's by at most BALANCE_LEADER_MAX_INDEX_GAP. That gap is
+ * estimated from store heartbeats taken at different times, with the leader's progress in
+ * between taken as its average rate since the previous run, so it is not a strict bound:
+ * a burst late in that time makes a caught-up target look behind, and a burst early in it
+ * hides a target that fell behind.
+ */
+ private boolean canTakeLeader(Metapb.ShardGroup group, long storeId,
+ Map<Integer, Map<Long, IndexReport>> indexReports,
+ Map<Integer, IndexReport> previousLeaderReports) throws
+ PDException {
+ // A PD that lost leadership during the run must not send what it planned as leader
+ if (!isLeader()) {
+ log.info("periodic leader balance keeps the leader of shard group {}, this PD is " +
+ "no longer the leader", group.getId());
+ return false;
+ }
+ if (!group.equals(storeService.getShardGroup(group.getId()))) {
+ log.info("periodic leader balance keeps the leader of shard group {}, the group " +
+ "changed after the check", group.getId());
+ return false;
+ }
+ Long leaderStore = getLeaderStore(group);
+ Map<Long, IndexReport> reports = indexReports.getOrDefault(group.getId(), Map.of());
+ IndexReport leader = leaderStore == null ? null : reports.get(leaderStore);
+ IndexReport target = reports.get(storeId);
+ if (leader == null || target == null) {
+ log.info("periodic leader balance keeps the leader of shard group {}, no applied " +
+ "index reported by store {} or leader store {}", group.getId(), storeId,
+ leaderStore);
+ return false;
+ }
+ // Stores send their heartbeats at different times, so a caught-up target reporting
+ // earlier than the leader looks behind by the writes in between. Estimate the leader's
+ // index at the time of the target's report, at the average rate it grew since the
+ // previous run. Without that rate, compare the indexes as reported.
+ long leaderIndex = leader.index();
+ IndexReport previous = previousLeaderReports.get(group.getId());
+ if (previous != null && previous.storeId() == leader.storeId() &&
+ previous.time() < leader.time() && previous.index() <= leader.index()) {
+ double rate = (double) (leader.index() - previous.index()) /
+ (leader.time() - previous.time());
+ leaderIndex += Math.round(rate * (target.time() - leader.time()));
Review Comment:
Fixed in e6c1bcb0d. `canTakeLeader` now skips a transfer when the target's or the leader Store's report is more than 60 s old (`BALANCE_LEADER_REPORT_MAX_AGE`, two Store heartbeats), before any backdating. I applied it to the leader's report as well, since an old leader report says as little about the leader's current index.
`testAutoBalancePartitionLeaderWithStaleReport` uses your case: after `warmUp(service, 70_000, 300_000)`, leaders at 100000 and store 3 at 75000 reported 250 s ago, store 3 now waits and store 2 takes both transfers. Store 3 at 95000 reported 50 s ago still takes two leaders, and a leader report 90 s old moves nothing.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +659,172 @@ 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. The plan uses the groups checked here; a group keeps
+ * its leader when this PD is no longer the leader, the group changed, or the target trails
+ * the leader by the time the transfer would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
+ Set<Long> activeStores = stores.stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ // shard group id -> store id -> the applied index from the store's last heartbeat
+ Map<Integer, Map<Long, IndexReport>> indexReports = new HashMap<>();
+ for (Metapb.Store store : stores) {
+ for (Metapb.RaftStats raftStats : store.getStats().getRaftStatsList()) {
+ indexReports.computeIfAbsent(raftStats.getPartitionId(), id -> new HashMap<>())
+ .put(store.getId(), new IndexReport(store.getId(),
+ raftStats.getCommittedIndex(),
+ store.getLastHeartbeat()));
+ }
+ }
+ // Plan with the leader reports of the previous run and keep this run's for the next,
+ // including runs that skip below, so a later run can tell the leader's write rate
+ Map<Integer, IndexReport> previousLeaderReports = new HashMap<>(lastLeaderIndexReports);
+ lastLeaderIndexReports.clear();
Review Comment:
Fixed in e6c1bcb0d along the lines of your first suggestion. Each group now keeps the latest leader report a run saw and the one before it from the same Store. When a run sees the same report as the previous run, it keeps both as they are, so the rate still comes from an older sample.
`testAutoBalancePartitionLeaderWithShortInterval` runs three times about 10 s apart against reports with fixed times: 7000 at t-40 s, then 10000 at t-10 s seen by the second and third runs. On the third run store 3 reports 8500 15 s before the leader, is caught up at 100 entries/s, and takes two leaders. With the old history that run had only the 10000 report, so it would compare raw indexes 1500 apart and store 3 would wait.
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/TaskScheduleServiceTest.java:
##########
@@ -75,6 +87,466 @@ public void testBalancePartitionLeader() throws PDException {
getStoreNodeService().getStoreInfoMeta().removeAll();
}
+ @Test
+ public void testAutoBalancePartitionLeaderWithinSkew() throws PDException {
Review Comment:
Added in e6c1bcb0d: `testAutoBalancePartitionLeaderWithSkewAtLimit` uses the default `max-skew` of 2. Leaders at 3/1/2 (skew 2) return no transfers, and 4/1/1 (skew 3) plan transfers, none of them to store 1. With `<`, the skew-2 layout would reach the planner and get transfers, so the first assertion would fail.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +659,172 @@ 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. The plan uses the groups checked here; a group keeps
+ * its leader when this PD is no longer the leader, the group changed, or the target trails
+ * the leader by the time the transfer would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
+ Set<Long> activeStores = stores.stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ // shard group id -> store id -> the applied index from the store's last heartbeat
+ Map<Integer, Map<Long, IndexReport>> indexReports = new HashMap<>();
+ for (Metapb.Store store : stores) {
+ for (Metapb.RaftStats raftStats : store.getStats().getRaftStatsList()) {
+ indexReports.computeIfAbsent(raftStats.getPartitionId(), id -> new HashMap<>())
+ .put(store.getId(), new IndexReport(store.getId(),
+ raftStats.getCommittedIndex(),
+ store.getLastHeartbeat()));
+ }
+ }
+ // Plan with the leader reports of the previous run and keep this run's for the next,
+ // including runs that skip below, so a later run can tell the leader's write rate
+ Map<Integer, IndexReport> previousLeaderReports = new HashMap<>(lastLeaderIndexReports);
+ lastLeaderIndexReports.clear();
+ for (Metapb.ShardGroup group : shardGroups) {
+ Long leaderStore = getLeaderStore(group);
+ IndexReport report = leaderStore == null ? null :
+ indexReports.getOrDefault(group.getId(), Map.of())
+ .get(leaderStore);
+ if (report != null) {
+ lastLeaderIndexReports.put(group.getId(), report);
+ }
+ }
+ 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 <= maxSkew) {
+ 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, maxSkew);
+ return balancePartitionLeader(false, shardGroups, (group, storeId) ->
Review Comment:
Done in e6c1bcb0d with your first option. The planner's target calculation moved into a `leaderTargets` helper that the planner still uses, and the periodic path skips the run, with one DEBUG line, when the leader counts already equal those targets. The planner keeps its "shard group size" INFO line but no longer logs the per-store shard counts there, so a skipped run logs nothing at INFO.
`testAutoBalancePartitionLeaderWithLeadersOnTargets` sets `max-skew` to 0 with seven groups on three stores. Leaders already at the planner's 2/2/3 return no transfers. Leaders at 3/2/2 have the same spread on other stores, so the planner runs and ends at 2/2/3.
##########
hugegraph-pd/docs/configuration.md:
##########
@@ -188,6 +191,9 @@ pd:
| `pd.patrol-interval` | Integer | `1800` | Interval (in seconds) for partition health patrol and automatic rebalancing. Lower values = more frequent checks. |
| `pd.initial-store-count` | Integer | `1` | Minimum number of Store nodes required for cluster to be operational. Set to expected initial store count. |
| `pd.initial-store-list` | String | `127.0.0.1:8500` | Comma-separated list of Store gRPC addresses to auto-activate on startup. Useful for bootstrapping. |
+| `pd.balance-leader-enable` | Boolean | `false` | Run partition leader balancing periodically on the PD leader. When off, leaders move back only through `/v1/task/balanceLeaders`. |
Review Comment:
Fixed in e6c1bcb0d: the row now says leaders move back "only through a manual call: `/v1/task/balanceLeaders`, `/v1/balanceLeaders` or the `balanceLeaders` gRPC call".
##########
hugegraph-pd/docs/configuration.md:
##########
@@ -196,6 +202,21 @@ pd:
- Production: `1800` (30 minutes) to reduce overhead
- Large clusters: `3600` (1 hour)
- `pd.initial-store-count`: Set to expected initial store count (e.g., `3` for 3 stores)
+- `pd.balance-leader-enable`: After a Store restarts, its raft groups keep the leaders they elected
+ on the surviving Stores. Enable this so PD moves leaders back without a manual
+ `balanceLeaders` call. A check is skipped while any shard group is not in the `Normal` state,
+ has sent no partition heartbeat in the last 30 seconds, or has a replica that is not a Leader
+ or Follower (a membership change in progress); while a Store that holds a shard is not sending
+ heartbeats; and while a split, move or shard balance task is running. A planned transfer is
Review Comment:
Fixed in e6c1bcb0d with your wording: "while a Store that holds a shard is not in the active-store list, which it leaves `store.keepAlive-timeout` after its last heartbeat". The same paragraph now also covers the 60 s report age and the target-match skip.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#issuecomment-6228195770
Thanks for the re-review. The `codecov/project` drop follows from which CI jobs ran. This PR only touches PD, so the CI preflight skipped `build-server`, `server-rocksdb` and `commons`, and Codecov compares the PD/Store-only upload with base 92af2f8 at 54.31%, which includes server coverage. #3354, a merged Store-only fix, failed the same check the same way: 34.80% (-19.51%), with `build-server` skipped. #3348 changed the server, ran those jobs and came in at 53.28% (-1.03%). `codecov/patch` passes here at 74.72% against a 54.31% target.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4355680030
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -143,6 +186,29 @@ public void init() {
}
}, 2, 30,
TimeUnit.SECONDS);
+ if (pdConfig.isBalanceLeaderEnable()) {
+ long interval = pdConfig.getBalanceLeaderInterval();
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (interval > 0 && maxSkew >= 0) {
+ balanceLeaderTask = executor.scheduleWithFixedDelay(() -> {
+ try {
+ autoBalancePartitionLeader();
Review Comment:
Fixed in f8c3a952f with your first option. Only the periodic path calls the planner with `immediately=false`; the REST, `/v1/balanceLeaders` and gRPC entries all pass `true`. So the planner's wait now uses `pd.balance-leader-interval` in place of the fixed 30 s, and the `BalanceLeaderInterval` constant is gone. A 10 s setting now plans at most every 10 s. The wait is still counted from the last plan of either kind, so the first periodic plan after a manual balance waits one interval.
Tests, both with a 1 s interval:
- `testAutoBalancePartitionLeaderFollowsInterval`: a first run moves four leaders, an immediate second run moves none, and a run 1.1 s later moves four again.
- `testAutoBalancePartitionLeaderWaitsAfterManualBalance`: a manual `balancePartitionLeader(true)` moves four, the periodic run right after it moves none, and a run 1.1 s later moves four.
With the fixed 30 s wait both fail on the run 1.1 s later, with no transfers. PD core suite: 213 tests, 0 failures, 2 skipped. The `pd.balance-leader-interval` row in `configuration.md` now says it also spaces out the plans.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
bitflicker64 commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4304095800
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +659,172 @@ 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. The plan uses the groups checked here; a group keeps
+ * its leader when this PD is no longer the leader, the group changed, or the target trails
+ * the leader by the time the transfer would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
+ Set<Long> activeStores = stores.stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ // shard group id -> store id -> the applied index from the store's last heartbeat
+ Map<Integer, Map<Long, IndexReport>> indexReports = new HashMap<>();
+ for (Metapb.Store store : stores) {
+ for (Metapb.RaftStats raftStats : store.getStats().getRaftStatsList()) {
+ indexReports.computeIfAbsent(raftStats.getPartitionId(), id -> new HashMap<>())
+ .put(store.getId(), new IndexReport(store.getId(),
+ raftStats.getCommittedIndex(),
+ store.getLastHeartbeat()));
+ }
+ }
+ // Plan with the leader reports of the previous run and keep this run's for the next,
+ // including runs that skip below, so a later run can tell the leader's write rate
+ Map<Integer, IndexReport> previousLeaderReports = new HashMap<>(lastLeaderIndexReports);
+ lastLeaderIndexReports.clear();
+ for (Metapb.ShardGroup group : shardGroups) {
+ Long leaderStore = getLeaderStore(group);
+ IndexReport report = leaderStore == null ? null :
+ indexReports.getOrDefault(group.getId(), Map.of())
+ .get(leaderStore);
+ if (report != null) {
+ lastLeaderIndexReports.put(group.getId(), report);
+ }
+ }
+ 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 <= maxSkew) {
+ 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, maxSkew);
+ return balancePartitionLeader(false, shardGroups, (group, storeId) ->
Review Comment:
Minor: The description says a layout whose best reachable spread is above `max-skew` makes every run plan and move nothing. The cost is log volume: each such run logs at INFO here, then the shard counts, the target counts and one "is leader, no need change" line per group.
Example: 12 groups with 3 replicas on 5 Stores holding 8/7/7/7/7 shards. The planner targets 3/3/2/2/2 leaders, so the skew stays 1, and with `max-skew: 0`, which the docs allow, that repeats every interval.
Please skip the run when the leader counts already match the planner's targets, or log the per-group line at DEBUG on the periodic path.
##########
hugegraph-pd/docs/configuration.md:
##########
@@ -188,6 +191,9 @@ pd:
| `pd.patrol-interval` | Integer | `1800` | Interval (in seconds) for partition health patrol and automatic rebalancing. Lower values = more frequent checks. |
| `pd.initial-store-count` | Integer | `1` | Minimum number of Store nodes required for cluster to be operational. Set to expected initial store count. |
| `pd.initial-store-list` | String | `127.0.0.1:8500` | Comma-separated list of Store gRPC addresses to auto-activate on startup. Useful for bootstrapping. |
+| `pd.balance-leader-enable` | Boolean | `false` | Run partition leader balancing periodically on the PD leader. When off, leaders move back only through `/v1/task/balanceLeaders`. |
Review Comment:
Minor: "leaders move back only through `/v1/task/balanceLeaders`" leaves out the other two manual entries.
`StoreAPI` also serves `/v1/balanceLeaders`, and the `balanceLeaders` gRPC call reaches `PDService.balanceLeaders`, which calls `balancePartitionLeader(true)`. The PR description lists all three.
Please name all three, or say "only through a manual balanceLeaders call".
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +659,172 @@ 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. The plan uses the groups checked here; a group keeps
+ * its leader when this PD is no longer the leader, the group changed, or the target trails
+ * the leader by the time the transfer would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
+ Set<Long> activeStores = stores.stream()
+ .map(Metapb.Store::getId)
+ .collect(Collectors.toSet());
+ // shard group id -> store id -> the applied index from the store's last heartbeat
+ Map<Integer, Map<Long, IndexReport>> indexReports = new HashMap<>();
+ for (Metapb.Store store : stores) {
+ for (Metapb.RaftStats raftStats : store.getStats().getRaftStatsList()) {
+ indexReports.computeIfAbsent(raftStats.getPartitionId(), id -> new HashMap<>())
+ .put(store.getId(), new IndexReport(store.getId(),
+ raftStats.getCommittedIndex(),
+ store.getLastHeartbeat()));
+ }
+ }
+ // Plan with the leader reports of the previous run and keep this run's for the next,
+ // including runs that skip below, so a later run can tell the leader's write rate
+ Map<Integer, IndexReport> previousLeaderReports = new HashMap<>(lastLeaderIndexReports);
+ lastLeaderIndexReports.clear();
Review Comment:
Important: With `pd.balance-leader-interval` below the 30 s Store heartbeat, two runs in a row can see the same leader report. This line then drops the older sample, so the next run finds `previous.time() == leader.time()` and compares raw indexes, as before 57bd3d19b.
I simulated ticks against 30 s heartbeats at 1000 entries/s with store 3 caught up, over 30 tick phases. With store 3 reporting 5 s before the leader, store 3 got no leaders on any planning run in 15 of 30 phases at 10 s, and in 10 of 30 at 15 s and at 20 s. At 30 and 300 s it never was. In general, an interval I below 30 s can defer a target that reports less than 30 - I s before the leader.
This only slows balancing. Please keep the latest sample older than the current report, or document a 30 s minimum for the interval.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4351246128
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -143,6 +186,29 @@ public void init() {
}
}, 2, 30,
TimeUnit.SECONDS);
+ if (pdConfig.isBalanceLeaderEnable()) {
+ long interval = pdConfig.getBalanceLeaderInterval();
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (interval > 0 && maxSkew >= 0) {
+ balanceLeaderTask = executor.scheduleWithFixedDelay(() -> {
+ try {
+ autoBalancePartitionLeader();
Review Comment:
⚠️ Important: This timer accepts any positive `pd.balance-leader-interval`, but the call reaches `balancePartitionLeader(false, ...)`, which still applies a fixed 30-second `BalanceLeaderInterval`. For example, with the documented valid value of 10 seconds, the callback checks every 10 seconds but eligible leader transfers are still throttled to at most once per 30 seconds. Please make the periodic planner honor this configured interval, or reject and document values below 30 seconds.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#issuecomment-6246789330
`codecov/project` at e6c1bcb0d: the failure comes from which CI jobs ran, not from this change, and the check is not required.
Codecov compares this head at 37.61% with base 92af2f8 at 54.31%. The base figure includes coverage from the server jobs. This PR touches only PD code and docs, so the CI preflight skipped `build-server`, `server-rocksdb` and `commons`, and the head upload has only the `hstore`, `pd` and `store` flags. The drop is the server coverage that was never uploaded. On the code this PR changes, `codecov/patch` passes at 90.48%, and the `pd` flag is at 51.13%; the 18 lines without full coverage are all in `TaskScheduleService`.
#3354, a merged Store-only PR, failed the same check the same way (34.80%, -19.51%) with the server jobs skipped. `.asf.yaml` lists no required status checks for `master` (`contexts: []`), so this result does not block the merge.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4300463681
##########
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:
Done in 3daaa200f. `autoBalancePartitionLeader()` now hands the groups it checked to the planner through a private `balancePartitionLeader` overload; the public method still reads them itself, so the manual path is unchanged. It is `synchronized` too, like the manual entry, so a manual call can't run between the check and the plan. Right before each transfer, `canTakeLeader` reads the group again and sends nothing if any field differs from the checked copy, including shard roles, `conf_ver` and a leader that moved in the meantime. `testAutoBalancePartitionLeaderWithChangeAfterCheck` turns store 3's shards into `None` members right after the balance reads the group list and expects no transfers.
That still leaves the time between the reread and the Store handling the instruction. JRaft 1.3.14 covers it on the Store side: `NodeImpl.transferLeadershipTo` refuses while the leader is changing its configuration and refuses a peer outside the current configuration, so a `None` or Learner shard can't receive leadership.
##########
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:
Fixed for the periodic path in 3daaa200f. A planned transfer now also needs the target's applied index to be within 1000 entries of the leader's. The indexes come from the `RaftStats` each Store sends with its store heartbeat (`PartitionStateMachine.committedIndex`, set in `onApply`), read through `getCommittedIndexStats()`. A target that reports no index is skipped, and so is a group without a leader shard. A skipped transfer still uses its slot in the plan, so that leader stays put until a later run and the other groups are balanced as usual.
The indexes can be up to 30 s old, since Store heartbeats run every 30 seconds, and the two reports aren't taken at the same moment. When the leader's report is newer, a caught-up follower can look behind, and the transfer is only deferred. When the follower's report is newer, a follower that fell behind after reporting can still pass. JRaft then times the transfer out after `electionTimeoutMs`, and the old leader returns to `STATE_LEADER` and accepts writes again. Closing that gap would need the Store to check replicator progress before accepting a transfer; I'd leave that out of this PR. The manual `balanceLeaders` path is unchanged.
`testAutoBalancePartitionLeaderWithLaggingFollower` puts all six leaders on store 1. With store 3 5000 entries behind, and again with store 3 reporting no index, there are two transfers, both to store 2. With store 3 500 behind there are four, two of them to store 3.
##########
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:
In 3daaa200f, `init()` schedules the periodic balance only when `pd.balance-leader-interval` is positive and `pd.balance-leader-max-skew` is 0 or more. Otherwise it logs one warning with both values and schedules nothing. The PR already handled a non-positive interval this way: a bad value turns the feature off and PD still starts. `autoBalancePartitionLeader()` also returns without planning on a negative skew, in case something calls it directly. `configuration.md` now gives the allowed range for both keys. `testAutoBalancePartitionLeaderWithNegativeSkew` sets -1 on a layout that moves four leaders at the default skew and expects no transfers.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
bitflicker64 commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4302468235
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +626,113 @@ 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. The plan uses the groups checked here; a group that
+ * changes before its transfer is sent, or whose target trails the leader, keeps its leader.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ 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 <= maxSkew) {
+ 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, maxSkew);
+ return balancePartitionLeader(false, shardGroups, true);
+ }
+
+ /**
+ * Checks a periodic transfer right before it is sent: the group must be unchanged since
+ * autoBalancePartitionLeader checked it, and the applied index the target's store last
+ * reported must trail the leader's by at most BALANCE_LEADER_MAX_INDEX_GAP.
+ */
+ private boolean canTakeLeader(Metapb.ShardGroup group, long storeId,
+ Map<Integer, Map<Long, Long>> committedIndexMap) throws
+ PDException {
+ if (!group.equals(storeService.getShardGroup(group.getId()))) {
+ log.info("periodic leader balance keeps the leader of shard group {}, the group " +
+ "changed after the check", group.getId());
+ return false;
+ }
+ Long leaderStore = group.getShardsList().stream()
+ .filter(shard -> shard.getRole() == Metapb.ShardRole.Leader)
+ .map(Metapb.Shard::getStoreId)
+ .findFirst().orElse(null);
+ Map<Long, Long> indexes = committedIndexMap.getOrDefault(group.getId(), Map.of());
+ Long leaderIndex = leaderStore == null ? null : indexes.get(leaderStore);
+ Long targetIndex = indexes.get(storeId);
+ if (leaderIndex == null || targetIndex == null ||
+ leaderIndex - targetIndex > BALANCE_LEADER_MAX_INDEX_GAP) {
Review Comment:
Important: The 1000-entry gap compares two applied indexes taken at different times, and the time difference is the same on every run, so under steady writes a follower that is caught up can be deferred on every run, not once. `getCommittedIndexStats()` reads the `RaftStats` from each Store's last store heartbeat. Stores send that heartbeat every 30 s with a fixed delay (`HgStoreEngineOptions.storeHBInterval = 30`, `HeartbeatService.doStoreHeartbeat`), and the default `pd.balance-leader-interval` of 300 s is a multiple of 30 s. When the target Store's last report is d seconds older than the leader Store's, a follower with no real lag still looks about (write rate x d) entries behind: at 100 entries/s on a partition and d = 15 s that is 1500, so the transfer is skipped. Both heartbeat timers and the 300 s timer drift only by RPC and run time, so the same pair sees the same offset run after run, and a restarted Store can get none of its leaders back for as long as the load lasts. Th
at is the case this feature is meant to fix, and the new tests only use indexes reported at the same moment.
Please account for when each index was reported before treating the gap as lag. One option is to use `lastHeartbeat`, which the active-store entry already carries, to allow for the leader's progress between the two reports. Another is to take follower progress from the leader itself, for example in its 5 s partition heartbeat. Please also add a test where the target's report is older than the leader's while the target is caught up.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#issuecomment-6192805853
## [Codecov](https://app.codecov.io/gh/apache/hugegraph/pull/3356?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `60.41667%` with `19 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 49.94%. Comparing base ([`92af2f8`](https://app.codecov.io/gh/apache/hugegraph/commit/92af2f89c320c3e6fa7c10765a061f3edadea319?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`f57f45d`](https://app.codecov.io/gh/apache/hugegraph/commit/f57f45da673a5450f17c1604c44eb8189a842b9d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 12 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/hugegraph/pull/3356?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...a/org/apache/hugegraph/pd/TaskScheduleService.java](https://app.codecov.io/gh/apache/hugegraph/pull/3356?src=pr&el=tree&filepath=hugegraph-pd%2Fhg-pd-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhugegraph%2Fpd%2FTaskScheduleService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVnZWdyYXBoLXBkL2hnLXBkLWNvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1Z2VncmFwaC9wZC9UYXNrU2NoZWR1bGVTZXJ2aWNlLmphdmE=) | 54.76% | [16 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/hugegraph/pull/3356?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3356 +/- ##
============================================
- Coverage 54.31% 49.94% -4.37%
+ Complexity 2250 847 -1403
============================================
Files 770 66 -704
Lines 67952 6185 -61767
Branches 9213 667 -8546
============================================
- Hits 36905 3089 -33816
+ Misses 27470 2848 -24622
+ Partials 3577 248 -3329
```
| [Flag](https://app.codecov.io/gh/apache/hugegraph/pull/3356/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [pd](https://app.codecov.io/gh/apache/hugegraph/pull/3356/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `49.94% <60.41%> (?)` | |
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/hugegraph/pull/3356?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4289857957
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +605,61 @@ 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 or a store
+ * that holds a shard is not active, because the target distribution would include a store
+ * 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());
+ 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) {
Review Comment:
⚠️ A shard group that loses quorum can stop sending PartitionStats, leaving its last PState_Normal in metadata; keepStoreAlive keeps each Store marked active until its TTL expires. With the 300-second defaults, the scheduled tick can run while both snapshots still look healthy and call the global planner on stale membership. Please also require fresh partition health evidence before balancing and add a no-quorum regression.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +605,61 @@ 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 or a store
+ * that holds a shard is not active, because the target distribution would include a store
+ * 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());
+ 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<>();
+ }
+ for (Metapb.Shard shard : group.getShardsList()) {
+ 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;
Review Comment:
⚠️ This counts every non-Leader shard as eligible, but StoreNodeService.reallocShards can add a shard with the default ShardRole.None, preserve the group state, and start CONF_CHANGE_TYPE_ADJUST. The existing planner treats any non-Leader shard as a transfer target, so a tick during that adjustment can request leadership transfer to a non-voting or Learner peer. Please skip pending membership changes/non-voting roles or restrict targets to eligible followers.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#issuecomment-6237398994
Thanks. Addressed in 67b2dc2da, without changing the algorithm.
- The 1000-entry check is now described as an estimated, best-effort gap: in the comments on `BALANCE_LEADER_MAX_INDEX_GAP` and `canTakeLeader`, in the skip log line ("at an estimated N"), in `configuration.md` and in the PR description. Each says a late burst can keep a caught-up Store waiting and an early burst can let a Store that fell behind through, with JRaft's transfer timeout bounding the second case.
- New tests, using your numbers:
- `testAutoBalancePartitionLeaderWithLateBurst`: 7000 to 10000 over 300 s, store 3 caught up at 7000 and reporting 15 s earlier. The estimate puts the leader at 9850, so store 3 waits and store 2 takes both transfers.
- `testAutoBalancePartitionLeaderWithEarlyBurst`: 7000 to 10000 over 30 s, store 3 at 8500 and reporting 15 s earlier. It passes and takes two leaders; the test comment notes it could have been 1500 behind.
- `testAutoBalancePartitionLeaderWithNewerFollowerReport`: store 3 reports 15 s after the leader. At 11400 it takes two leaders, at 10000 (1500 behind the estimated 11500) it waits, and with no earlier run the same reports compare as equal.
- History wording: the field comment now says the entry is used only while the same Store leads the group, is cleared by a run that finds this PD is not the leader, and is not tied to a leadership term. A PD that loses and regains leadership, or a leader that moves away and back, between two runs keeps the entry, and the rate is averaged over that time. The docs say "no usable earlier sample from the same leader Store" where they said "first run".
PD core suite: 165 tests, 0 failures, 2 skipped.
The `pd_store / hstore` failure on 57bd3d19b is in CI setup: the start script appends `usePD=true` on every start, so `init-store.sh` reads `[true, true]` and exits before any API test runs. Other PRs failed the same way today, and #3371 fixes the script.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4332682186
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +630,257 @@ 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 and does
+ * not already match the planner's targets, so a balanced cluster sees no transfers. Skipped while a shard group is not Normal, has no
Review Comment:
Fixed in 7119a6fe4: the paragraph is rewrapped at the width the rest of the file uses, with the wording unchanged. No line this PR adds is over 120 characters now.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#issuecomment-6197343559
Thanks, fixed in bc694188a. `autoBalancePartitionLeader()` now returns before planning when any shard group is not `PState_Normal`, and logs the group and its state. A group in `PState_None` (no stats reported yet) also skips the run.
`testAutoBalancePartitionLeaderWithWarnGroup` keeps Stores 1-3 active with every leader on Store 1 and group 0 in `Warn`, and expects no transfers. The same layout with group 0 back to `Normal` moves four leaders. Adding it showed that another test in the suite leaves shard group 6 behind in `PState_Offline`, so the helper now sets existing groups aside for each case and restores them afterwards.
PD core suite: 154 tests, 0 failures, 2 skipped. The configuration doc lists the new skip condition.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4305216557
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/TaskScheduleServiceTest.java:
##########
@@ -75,6 +87,466 @@ public void testBalancePartitionLeader() throws PDException {
getStoreNodeService().getStoreInfoMeta().removeAll();
}
+ @Test
+ public void testAutoBalancePartitionLeaderWithinSkew() throws PDException {
Review Comment:
⚠️ Important: These cases cover skew 0 (`2/2/2`) and skew 6 (`6/0/0`) with `maxSkew=2`, but do not pin the threshold. An accidental change from `<=` to `<` would still pass while triggering transfers at exactly the configured limit. Please add cases for skew 2 (skip) and skew 3 (balance) with `maxSkew=2`.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "bitflicker64 (via GitHub)" <gi...@apache.org>.
bitflicker64 commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4328416898
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +630,257 @@ 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 and does
+ * not already match the planner's targets, so a balanced cluster sees no transfers. Skipped while a shard group is not Normal, has no
Review Comment:
Minor: This Javadoc line is 138 characters. The project limit is 120 (`max_line_length = 120` in `.editorconfig`, `LineLength` max 120 in `style/checkstyle.xml`), and the rest of this file wraps at 100. It looks like the sentence added in e6c1bcb0d was not rewrapped. Please wrap it, for example:
```suggestion
* not already match the planner's targets, so a balanced cluster sees no transfers.
* Skipped while a shard group is not Normal, has no
```
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4339724925
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +630,258 @@ 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 and does not
+ * already match the planner's targets, 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. The plan uses the
+ * groups checked here; a group keeps its leader when this PD is no longer the leader, the
+ * group changed, the target or leader store has not reported for
+ * BALANCE_LEADER_REPORT_MAX_AGE, or the target trails the leader by the time the transfer
+ * would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ earlierLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
Review Comment:
⚠️ Important: `getActiveStores()` reads TTL-backed snapshots. When an online Store changes to `Exiting`, `StoreNodeService.updateStore()` leaves its TTL entry in place and the Exiting heartbeat path does not refresh it; `storeTurnoff()` can rebuild shard membership from that same list, and `reallocShards()` can add the Exiting Store back as a shard. Once that group is Normal and the shard is a Follower, this periodic path can select it and move a leader back to a Store being drained. Please exclude Stores whose current persisted state is `Exiting` from periodic targets and recheck that state 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: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4303678270
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +626,113 @@ 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. The plan uses the groups checked here; a group that
+ * changes before its transfer is sent, or whose target trails the leader, keeps its leader.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ 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 <= maxSkew) {
+ 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, maxSkew);
+ return balancePartitionLeader(false, shardGroups, true);
+ }
+
+ /**
+ * Checks a periodic transfer right before it is sent: the group must be unchanged since
+ * autoBalancePartitionLeader checked it, and the applied index the target's store last
+ * reported must trail the leader's by at most BALANCE_LEADER_MAX_INDEX_GAP.
+ */
+ private boolean canTakeLeader(Metapb.ShardGroup group, long storeId,
+ Map<Integer, Map<Long, Long>> committedIndexMap) throws
+ PDException {
+ if (!group.equals(storeService.getShardGroup(group.getId()))) {
+ log.info("periodic leader balance keeps the leader of shard group {}, the group " +
+ "changed after the check", group.getId());
+ return false;
+ }
+ Long leaderStore = group.getShardsList().stream()
+ .filter(shard -> shard.getRole() == Metapb.ShardRole.Leader)
+ .map(Metapb.Shard::getStoreId)
+ .findFirst().orElse(null);
+ Map<Long, Long> indexes = committedIndexMap.getOrDefault(group.getId(), Map.of());
+ Long leaderIndex = leaderStore == null ? null : indexes.get(leaderStore);
+ Long targetIndex = indexes.get(storeId);
+ if (leaderIndex == null || targetIndex == null ||
+ leaderIndex - targetIndex > BALANCE_LEADER_MAX_INDEX_GAP) {
Review Comment:
A correction to my reply above: "the first run after this PD becomes leader" was too strong. The history is keyed by group and leader Store, not by leadership term, and only a run that finds this PD is not the leader clears it. If PD leadership is lost and regained, or the group's leader moves away and back, between two runs, the old entry stays and the rate is averaged over that whole time. 67b2dc2da says so in the code and docs.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4303294367
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +626,113 @@ 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. The plan uses the groups checked here; a group that
+ * changes before its transfer is sent, or whose target trails the leader, keeps its leader.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
Review Comment:
Fixed in 57bd3d19b. `canTakeLeader` now checks `isLeader()` first, right before each periodic transfer goes to `PartitionService.transferLeader()`, so a PD that lost leadership during the run sends nothing it planned. `testAutoBalancePartitionLeaderAfterLosingLeadership` uses a service whose `isLeader()` is true only for the entry check: the run sends no transfers, and the test asserts the planned transfers reached the recheck. `isLeader()` is `protected` now so the test can override it.
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/TaskScheduleServiceTest.java:
##########
@@ -75,6 +83,217 @@ public void testBalancePartitionLeader() throws PDException {
getStoreNodeService().getStoreInfoMeta().removeAll();
}
+ @Test
+ public void testAutoBalancePartitionLeaderWithinSkew() throws PDException {
+ // leaders spread 2/2/2 over the three stores
+ var rst = autoBalance(new long[]{1, 2, 3, 1, 2, 3}, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderOverSkew() throws PDException {
+ // every leader on store 1, as after stores 2 and 3 restarted
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2, 3);
+ assertEquals(4, rst.size());
+ assertFalse(rst.containsValue(1L));
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithInactiveStore() throws PDException {
+ // store 3 holds shards but sends no heartbeat, so it cannot take a leader
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithWarnGroup() throws PDException {
+ // store 3 just stopped: still in the active set, but its groups are already Warn
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1},
+ g -> g.toBuilder().setState(Metapb.PartitionState.PState_Warn)
+ .build(), 0, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // once the group is Normal again, the same imbalance is balanced
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0, 1, 2, 3);
+ assertEquals(4, rst.size());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithoutQuorum() throws PDException {
+ // group 0 lost its quorum: no partition heartbeat for a minute, while its last state
+ // is still Normal and every store is still in the active set
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 60 * 1000, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // group 0 never reported stats at all
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, -1, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // a heartbeat within the window: the same imbalance is balanced
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 10 * 1000, 1, 2, 3);
+ assertEquals(4, rst.size());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithPendingMember() throws PDException {
+ // reallocShards added the shard on store 3 and left its role None
+ for (var role : List.of(Metapb.ShardRole.None, Metapb.ShardRole.Learner)) {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1},
+ g -> g.toBuilder().setShards(2, g.getShards(2).toBuilder()
+ .setRole(role))
+ .build(), 0, 1, 2, 3);
+ assertTrue(role.name(), rst.isEmpty());
+ }
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithChangeAfterCheck() throws PDException {
+ // reallocShards turns store 3's shards into new None members right after the check
+ // read the groups: the plan is made from the checked groups and no transfer is sent
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> 10_000,
+ g -> g.toBuilder().setShards(2, g.getShards(2).toBuilder()
+ .setRole(Metapb.ShardRole.None))
+ .build(), 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithLaggingFollower() throws PDException {
+ // store 3's replicas trail the leaders by 5000 entries or report no index: its share
+ // of the leaders stays on store 1 and only store 2 takes leaders
+ for (long store3Index : new long[]{5_000, -1}) {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> storeId == 3 ? store3Index : 10_000,
+ null, 1, 2, 3);
+ assertEquals(2, rst.size());
+ assertTrue(rst.values().stream().allMatch(storeId -> storeId == 2L));
+ }
+
+ // 500 entries behind is close enough to take a leader
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> storeId == 3 ? 9_500 : 10_000,
+ null, 1, 2, 3);
+ assertEquals(4, rst.size());
+ assertTrue(rst.containsValue(3L));
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithNegativeSkew() throws PDException {
+ // with the default skew this layout moves four leaders; -1 must not run a plan
+ getPdConfig().setBalanceLeaderMaxSkew(-1);
+ try {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ } finally {
+ getPdConfig().setBalanceLeaderMaxSkew(2);
+ }
+ }
+
+ private Map<Integer, Long> autoBalance(long[] leaderStores, long... activeStores) throws
+ PDException {
+ return autoBalance(leaderStores, g -> g, 0, activeStores);
+ }
+
+ private Map<Integer, Long> autoBalance(long[] leaderStores,
+ UnaryOperator<Metapb.ShardGroup> group0,
+ long group0StatsAge,
+ long... activeStores) throws PDException {
+ return autoBalance(leaderStores, group0, group0StatsAge, (storeId, groupId) -> 10_000,
+ null, activeStores);
+ }
+
+ /**
+ * Stores shard groups 0..n-1 on stores 1, 2 and 3 with the leader of group i on
+ * leaderStores[i], all Normal with fresh partition stats, then applies group0 to group 0
+ * and ages its stats by group0StatsAge ms (-1: no stats). Marks activeStores alive, each
+ * reporting appliedIndex(store, group) for every group (negative: not reported), and runs
+ * one periodic balance. A non-null afterRead is applied to every stored group right after
+ * the balance reads the group list, like a change landing between check and transfer.
+ */
+ private Map<Integer, Long> autoBalance(long[] leaderStores,
+ UnaryOperator<Metapb.ShardGroup> group0,
+ long group0StatsAge,
+ ToLongBiFunction<Long, Integer> appliedIndex,
+ UnaryOperator<Metapb.ShardGroup> afterRead,
+ long... activeStores) throws PDException {
+ var storeMeta = getStoreNodeService().getStoreInfoMeta();
+ var partitionMeta = MetadataFactory.newPartitionMeta(getPdConfig());
+ // Set aside groups other tests in the suite left behind, so only ours are planned
+ var otherGroups = storeMeta.getShardGroups();
+ for (var group : otherGroups) {
+ storeMeta.deleteShardGroup(group.getId());
+ }
+ var partitions = new ArrayList<Metapb.Partition>();
+ long now = System.currentTimeMillis();
+ for (int i = 0; i < leaderStores.length; i++) {
+ var group = genShardGroup(i, leaderStores[i], Metapb.PartitionState.PState_Normal);
+ long statsAge = 0;
+ if (i == 0) {
+ group = group0.apply(group);
+ statsAge = group0StatsAge;
+ }
+ storeMeta.updateShardGroup(group);
+ if (statsAge >= 0) {
+ partitionMeta.updatePartitionStats(Metapb.PartitionStats.newBuilder()
+ .setId(i)
+ .setTimestamp(
+ now - statsAge)
+ .build());
+ }
+ partitions.add(genPartition(i));
+ }
+ for (long storeId = 1; storeId <= 3; storeId++) {
+ var stats = Metapb.StoreStats.newBuilder();
+ for (int i = 0; i < leaderStores.length; i++) {
+ long index = appliedIndex.applyAsLong(storeId, i);
+ if (index >= 0) {
+ stats.addRaftStats(Metapb.RaftStats.newBuilder().setPartitionId(i)
+ .setCommittedIndex(index));
+ }
+ }
+ var store = Metapb.Store.newBuilder().setId(storeId).setStats(stats).build();
+ if (Arrays.stream(activeStores).anyMatch(id -> id == store.getId())) {
+ storeMeta.keepStoreAlive(store);
+ } else {
+ storeMeta.removeActiveStore(store);
+ }
+ }
+ getPdConfig().getPartition().setShardCount(3);
+ getPartitionService().updatePartition(partitions);
+ try {
+ StoreNodeService storeService = getStoreNodeService();
+ if (afterRead != null) {
+ storeService = new StoreNodeService(getPdConfig()) {
+ @Override
+ public List<Metapb.ShardGroup> getShardGroups() throws PDException {
+ var groups = super.getShardGroups();
+ for (var group : groups) {
+ storeMeta.updateShardGroup(afterRead.apply(group));
+ }
+ return groups;
+ }
+ };
+ }
+ // A fresh service, so a balance run by another test does not hit the rate limit
+ var service = new TaskScheduleService(getPdConfig(), storeService,
+ getPartitionService());
+ return service.autoBalancePartitionLeader();
Review Comment:
Added in 57bd3d19b: `testInitSchedulesLeaderBalance` calls `init()` on a service whose `autoBalancePartitionLeader()` only counts down a latch. Disabled, interval 0, interval -5 and skew -1 leave the timer unscheduled. Skew 0 and skew 2 with a 1 s interval schedule it, and the latch must fire within 10 s. `init()` keeps the scheduled future and `isLeaderBalanceScheduled()` reports whether there is one, which is what the test checks for each case.
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +626,113 @@ 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. The plan uses the groups checked here; a group that
+ * changes before its transfer is sent, or whose target trails the leader, keeps its leader.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ 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 <= maxSkew) {
+ 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, maxSkew);
+ return balancePartitionLeader(false, shardGroups, true);
+ }
+
+ /**
+ * Checks a periodic transfer right before it is sent: the group must be unchanged since
+ * autoBalancePartitionLeader checked it, and the applied index the target's store last
+ * reported must trail the leader's by at most BALANCE_LEADER_MAX_INDEX_GAP.
+ */
+ private boolean canTakeLeader(Metapb.ShardGroup group, long storeId,
+ Map<Integer, Map<Long, Long>> committedIndexMap) throws
+ PDException {
+ if (!group.equals(storeService.getShardGroup(group.getId()))) {
+ log.info("periodic leader balance keeps the leader of shard group {}, the group " +
+ "changed after the check", group.getId());
+ return false;
+ }
+ Long leaderStore = group.getShardsList().stream()
+ .filter(shard -> shard.getRole() == Metapb.ShardRole.Leader)
+ .map(Metapb.Shard::getStoreId)
+ .findFirst().orElse(null);
+ Map<Long, Long> indexes = committedIndexMap.getOrDefault(group.getId(), Map.of());
+ Long leaderIndex = leaderStore == null ? null : indexes.get(leaderStore);
+ Long targetIndex = indexes.get(storeId);
+ if (leaderIndex == null || targetIndex == null ||
+ leaderIndex - targetIndex > BALANCE_LEADER_MAX_INDEX_GAP) {
Review Comment:
Done in 57bd3d19b, along the lines of your first option. The check now reads each Store's `lastHeartbeat` along with its `RaftStats`, and every periodic run keeps the leader Store's report for each group, including runs that skip. On the next run the leader's write rate is (index now - index then) / (time now - time then), and PD moves the leader's index to the time of the target's report at that rate before comparing. A target whose report is d seconds older is compared with the leader's estimated index d seconds earlier, so the fixed heartbeat phase no longer shows up as lag on every run. Until a run has seen a rate (the first run after this PD becomes leader, or after the group's leader moved), the indexes are compared as reported, so a caught-up Store can wait one extra run. The rate is an average over the balance interval, so bursty writes still add some error; the 1000-entry gap leaves room for part of it.
`testAutoBalancePartitionLeaderWithOlderFollowerReport`: the leaders on store 1 apply 100 entries/s and store 3's heartbeat is 15 s older, so its caught-up replicas report 8500 against 10000. With no earlier run, store 3 waits: two transfers, both to store 2. After a run 30 s earlier that saw the leaders at 7000, the same reports give four transfers, two of them to store 3. A report of 5000 from store 3 at that rate is 3500 behind and still waits.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4301480191
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +626,113 @@ 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. The plan uses the groups checked here; a group that
+ * changes before its transfer is sent, or whose target trails the leader, keeps its leader.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
Review Comment:
⚠️ Important: `isLeader()` is checked only at the start of this periodic pass. If leadership changes while this method reads shard groups and index stats, it still enters the planner; `canTakeLeader()` rechecks the group and indexes but not PD leadership, and `PartitionService.transferLeader()` forwards through this PD's local pulse subject. Please recheck leadership before dispatch so a former PD cannot send stale transfer instructions.
##########
hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/TaskScheduleServiceTest.java:
##########
@@ -75,6 +83,217 @@ public void testBalancePartitionLeader() throws PDException {
getStoreNodeService().getStoreInfoMeta().removeAll();
}
+ @Test
+ public void testAutoBalancePartitionLeaderWithinSkew() throws PDException {
+ // leaders spread 2/2/2 over the three stores
+ var rst = autoBalance(new long[]{1, 2, 3, 1, 2, 3}, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderOverSkew() throws PDException {
+ // every leader on store 1, as after stores 2 and 3 restarted
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2, 3);
+ assertEquals(4, rst.size());
+ assertFalse(rst.containsValue(1L));
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithInactiveStore() throws PDException {
+ // store 3 holds shards but sends no heartbeat, so it cannot take a leader
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithWarnGroup() throws PDException {
+ // store 3 just stopped: still in the active set, but its groups are already Warn
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1},
+ g -> g.toBuilder().setState(Metapb.PartitionState.PState_Warn)
+ .build(), 0, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // once the group is Normal again, the same imbalance is balanced
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0, 1, 2, 3);
+ assertEquals(4, rst.size());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithoutQuorum() throws PDException {
+ // group 0 lost its quorum: no partition heartbeat for a minute, while its last state
+ // is still Normal and every store is still in the active set
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 60 * 1000, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // group 0 never reported stats at all
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, -1, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+
+ // a heartbeat within the window: the same imbalance is balanced
+ rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 10 * 1000, 1, 2, 3);
+ assertEquals(4, rst.size());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithPendingMember() throws PDException {
+ // reallocShards added the shard on store 3 and left its role None
+ for (var role : List.of(Metapb.ShardRole.None, Metapb.ShardRole.Learner)) {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1},
+ g -> g.toBuilder().setShards(2, g.getShards(2).toBuilder()
+ .setRole(role))
+ .build(), 0, 1, 2, 3);
+ assertTrue(role.name(), rst.isEmpty());
+ }
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithChangeAfterCheck() throws PDException {
+ // reallocShards turns store 3's shards into new None members right after the check
+ // read the groups: the plan is made from the checked groups and no transfer is sent
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> 10_000,
+ g -> g.toBuilder().setShards(2, g.getShards(2).toBuilder()
+ .setRole(Metapb.ShardRole.None))
+ .build(), 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithLaggingFollower() throws PDException {
+ // store 3's replicas trail the leaders by 5000 entries or report no index: its share
+ // of the leaders stays on store 1 and only store 2 takes leaders
+ for (long store3Index : new long[]{5_000, -1}) {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> storeId == 3 ? store3Index : 10_000,
+ null, 1, 2, 3);
+ assertEquals(2, rst.size());
+ assertTrue(rst.values().stream().allMatch(storeId -> storeId == 2L));
+ }
+
+ // 500 entries behind is close enough to take a leader
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, g -> g, 0,
+ (storeId, groupId) -> storeId == 3 ? 9_500 : 10_000,
+ null, 1, 2, 3);
+ assertEquals(4, rst.size());
+ assertTrue(rst.containsValue(3L));
+ }
+
+ @Test
+ public void testAutoBalancePartitionLeaderWithNegativeSkew() throws PDException {
+ // with the default skew this layout moves four leaders; -1 must not run a plan
+ getPdConfig().setBalanceLeaderMaxSkew(-1);
+ try {
+ var rst = autoBalance(new long[]{1, 1, 1, 1, 1, 1}, 1, 2, 3);
+ assertTrue(rst.isEmpty());
+ } finally {
+ getPdConfig().setBalanceLeaderMaxSkew(2);
+ }
+ }
+
+ private Map<Integer, Long> autoBalance(long[] leaderStores, long... activeStores) throws
+ PDException {
+ return autoBalance(leaderStores, g -> g, 0, activeStores);
+ }
+
+ private Map<Integer, Long> autoBalance(long[] leaderStores,
+ UnaryOperator<Metapb.ShardGroup> group0,
+ long group0StatsAge,
+ long... activeStores) throws PDException {
+ return autoBalance(leaderStores, group0, group0StatsAge, (storeId, groupId) -> 10_000,
+ null, activeStores);
+ }
+
+ /**
+ * Stores shard groups 0..n-1 on stores 1, 2 and 3 with the leader of group i on
+ * leaderStores[i], all Normal with fresh partition stats, then applies group0 to group 0
+ * and ages its stats by group0StatsAge ms (-1: no stats). Marks activeStores alive, each
+ * reporting appliedIndex(store, group) for every group (negative: not reported), and runs
+ * one periodic balance. A non-null afterRead is applied to every stored group right after
+ * the balance reads the group list, like a change landing between check and transfer.
+ */
+ private Map<Integer, Long> autoBalance(long[] leaderStores,
+ UnaryOperator<Metapb.ShardGroup> group0,
+ long group0StatsAge,
+ ToLongBiFunction<Long, Integer> appliedIndex,
+ UnaryOperator<Metapb.ShardGroup> afterRead,
+ long... activeStores) throws PDException {
+ var storeMeta = getStoreNodeService().getStoreInfoMeta();
+ var partitionMeta = MetadataFactory.newPartitionMeta(getPdConfig());
+ // Set aside groups other tests in the suite left behind, so only ours are planned
+ var otherGroups = storeMeta.getShardGroups();
+ for (var group : otherGroups) {
+ storeMeta.deleteShardGroup(group.getId());
+ }
+ var partitions = new ArrayList<Metapb.Partition>();
+ long now = System.currentTimeMillis();
+ for (int i = 0; i < leaderStores.length; i++) {
+ var group = genShardGroup(i, leaderStores[i], Metapb.PartitionState.PState_Normal);
+ long statsAge = 0;
+ if (i == 0) {
+ group = group0.apply(group);
+ statsAge = group0StatsAge;
+ }
+ storeMeta.updateShardGroup(group);
+ if (statsAge >= 0) {
+ partitionMeta.updatePartitionStats(Metapb.PartitionStats.newBuilder()
+ .setId(i)
+ .setTimestamp(
+ now - statsAge)
+ .build());
+ }
+ partitions.add(genPartition(i));
+ }
+ for (long storeId = 1; storeId <= 3; storeId++) {
+ var stats = Metapb.StoreStats.newBuilder();
+ for (int i = 0; i < leaderStores.length; i++) {
+ long index = appliedIndex.applyAsLong(storeId, i);
+ if (index >= 0) {
+ stats.addRaftStats(Metapb.RaftStats.newBuilder().setPartitionId(i)
+ .setCommittedIndex(index));
+ }
+ }
+ var store = Metapb.Store.newBuilder().setId(storeId).setStats(stats).build();
+ if (Arrays.stream(activeStores).anyMatch(id -> id == store.getId())) {
+ storeMeta.keepStoreAlive(store);
+ } else {
+ storeMeta.removeActiveStore(store);
+ }
+ }
+ getPdConfig().getPartition().setShardCount(3);
+ getPartitionService().updatePartition(partitions);
+ try {
+ StoreNodeService storeService = getStoreNodeService();
+ if (afterRead != null) {
+ storeService = new StoreNodeService(getPdConfig()) {
+ @Override
+ public List<Metapb.ShardGroup> getShardGroups() throws PDException {
+ var groups = super.getShardGroups();
+ for (var group : groups) {
+ storeMeta.updateShardGroup(afterRead.apply(group));
+ }
+ return groups;
+ }
+ };
+ }
+ // A fresh service, so a balance run by another test does not hit the rate limit
+ var service = new TaskScheduleService(getPdConfig(), storeService,
+ getPartitionService());
+ return service.autoBalancePartitionLeader();
Review Comment:
⚠️ Important: Every added case calls `autoBalancePartitionLeader()` directly through this helper; none exercises `init()` or asserts that the timer is registered only for valid enabled configuration. A regression in the new schedule/config gate would leave the feature inactive (or schedule it under invalid settings) while these tests pass. Please add deterministic coverage for disabled, valid, and invalid configuration.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
Re: [PR] feat(pd): add optional periodic partition leader balancing [hugegraph]
Posted by "arshilkxwork (via GitHub)" <gi...@apache.org>.
arshilkxwork commented on code in PR #3356:
URL: https://github.com/apache/hugegraph/pull/3356#discussion_r4340777109
##########
hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java:
##########
@@ -582,6 +630,258 @@ 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 and does not
+ * already match the planner's targets, 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. The plan uses the
+ * groups checked here; a group keeps its leader when this PD is no longer the leader, the
+ * group changed, the target or leader store has not reported for
+ * BALANCE_LEADER_REPORT_MAX_AGE, or the target trails the leader by the time the transfer
+ * would be sent.
+ *
+ * @return shard group id -> the store id its leader was moved to; empty when skipped
+ */
+ public synchronized Map<Integer, Long> autoBalancePartitionLeader() throws PDException {
+ // init() does not schedule this with a negative skew, which would balance every run
+ int maxSkew = pdConfig.getBalanceLeaderMaxSkew();
+ if (!isLeader() || maxSkew < 0) {
+ lastLeaderIndexReports.clear();
+ earlierLeaderIndexReports.clear();
+ return new HashMap<>();
+ }
+
+ List<Metapb.ShardGroup> shardGroups = storeService.getShardGroups();
+ if (shardGroups.isEmpty()) {
+ return new HashMap<>();
+ }
+
+ List<Metapb.Store> stores = storeService.getActiveStores();
Review Comment:
Fixed in 2818f21a0. Each periodic run now reads the persisted Store states (`StoreNodeService.getStores()`, not the TTL-backed active list) and only a Store whose state is `Up` can be a target. Right before each transfer, `canTakeLeader` reads the target's persisted state again and keeps the leader where it is unless it is still `Up`. A Store PD has set to `Exiting` is excluded even while its active entry lives on and `reallocShards` has added it back as a Follower. A deferred transfer keeps its slot in the plan, as with the other per-transfer checks, so the other groups are still balanced.
Tests:
- `testAutoBalancePartitionLeaderWithExitingStore`: all six leaders on store 1, store 3 still active but persisted as `Exiting`, then as `Offline`. Store 3 gets no leader and store 2 takes two.
- `testAutoBalancePartitionLeaderWithStoreExitingAfterRead`: store 3 is `Up` when the run reads the Store list and turns `Exiting` before the transfers are sent. The recheck keeps its two leaders on store 1.
Both fail without the change, with four transfers including two to store 3. PD core suite: 211 tests, 0 failures, 2 skipped. `configuration.md` lists the new condition.
--
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@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org