bitflicker64 commented on code in PR #3234:
URL: https://github.com/apache/hugegraph/pull/3234#discussion_r4092684606
##########
hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java:
##########
@@ -433,6 +411,132 @@ public Status changePeers(List<String> peers, final
Closure done) {
return removeSelf ? HgRaftError.TASK_CONTINUE.toStatus() :
HgRaftError.OK.toStatus();
}
+ /**
+ * Wait until the replicator of every peer is in Replicate state (snapshot
installed).
+ *
+ * @return false if the peers did not catch up within 600s
+ */
+ private boolean waitForReplicate(List<String> peers) {
+ //todo Each learner will wait for 1s, if another one is not
sync.Consider using
+ // countdownLatch
+ boolean allLearnerSnapshotOk = false;
+ long current = System.currentTimeMillis();
+ while (!allLearnerSnapshotOk) {
+ boolean snapshotOk = true;
+ for (var peerId : peers) {
+ var state = getReplicatorState(JRaftUtils.getPeerId(peerId));
+ log.info("Raft {}, peer:{}, replicate state:{}", getGroupId(),
peerId, state);
+ if (state != Replicator.State.Replicate) {
+ snapshotOk = false;
+ }
+ }
+ allLearnerSnapshotOk = snapshotOk;
+
+ if (!allLearnerSnapshotOk) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ log.warn("Raft {} sleep when check learner snapshot",
getGroupId());
+ }
+ }
+ if (System.currentTimeMillis() - current > 600 * 1000) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * PD's shard list and the raft configuration have the same endpoints, but
PD names another
+ * store id at one of them: a Store rebuilt with an empty disk at the same
raft address
+ * registers under a new id. The raft configuration has nothing to change,
so:
+ * 1. Create the raft node on that endpoint, the leader then installs a
snapshot into it.
+ * 2. Take PD's store ids into the local shard group and report it to PD,
otherwise the
+ * partition heartbeat writes the old id back to PD.
+ * 3. Wait for the peer to catch up.
+ *
+ * @param shards shard list from PD
+ * @return OK when nothing changed or the peers caught up, TASK_CONTINUE
to retry
+ */
+ private Status syncShardIdentities(List<Metapb.Shard> shards) {
+ Map<String, Long> pdIds = partitionManager.shardIdsByEndpoint(shards);
+ Map<String, Long> localIds =
+
partitionManager.shardIdsByEndpoint(shardGroup.getMetaPbShard());
+ String self =
raftNode.getNodeId().getPeerId().getEndpoint().toString();
+ List<String> changed = changedShardEndpoints(pdIds, localIds,
+
RaftUtils.getAllEndpoints(raftNode), self);
+ if (changed.isEmpty()) {
+ return HgRaftError.OK.toStatus();
+ }
+ log.info("Raft {} store id changed at raft address {}, local {}, pd
{}",
+ getGroupId(), changed, localIds, pdIds);
+
+ doSnapshot(status -> log.info("Raft {} snapshot before create raft
node, result:{}",
+ getGroupId(), status));
+ for (String peer : changed) {
+ FutureClosure closure = new FutureClosure();
+ storeEngine.getHgCmdClient().createRaftNode(peer,
partitionManager.getPartitionList(
+ getGroupId()), getCurrentConf(), closure);
+ Status status = closure.get();
+ if (!status.isOk()) {
+ log.info("Raft {} createRaftNode, peer:{}, reason:{}",
getGroupId(), peer,
+ status.getErrorMsg());
+ return HgRaftError.TASK_CONTINUE.toStatus();
Review Comment:
Minor: When `createRaftNode` fails here, the method returns `TASK_CONTINUE`,
and `doChangeShard` re-queues the task straight away through
`addRaftTask(SYNC_PARTITION_TASK)`. Before that, local ids are not updated yet,
so the retry finds the same changed endpoint and calls `doSnapshot` again on
line 474. A single failure takes about 1.5 s: `HgCmdClient.tryWithThrowable`
retries 5 times with a 100/200/300/400/500 ms backoff. Each retry also appends
a raft entry, and `setSnapshotLogIndexMargin` is commented out in
`PartitionEngine.init`, so jraft does not skip these snapshots. If the rebuilt
Store goes down again after PD has moved the group to its new id (for example a
crash-looping Pod), the leader of every affected group takes a snapshot and
appends a log entry about every 1.5 to 2 s until the Store is back.
`changePeers` handles the same `createRaftNode` failure at line 329 by
returning the RPC status (code -1), which is not `TASK_CONTINUE` and does not
re-queue. Please do the sa
me here, or at least call `doSnapshot` only once per repair (for example after
`createRaftNode` succeeds, or only when `getReplicatorState` shows the peer is
not already installing a snapshot), so a peer that stays down does not trigger
a retry loop that snapshots on every pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
