bitflicker64 commented on code in PR #3234:
URL: https://github.com/apache/hugegraph/pull/3234#discussion_r4094085333
##########
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:
Fixed in 97f5b013. A failed `createRaftNode` now returns the RPC status, as
`changePeers` does at line 329, so `doChangeShard` no longer re-queues it. The
task is already stored by the `SYNC_PARTITION_TASK` handler, and
`ReplicatorStateListener` runs it again when the endpoint comes back online.
The snapshot is taken once, after the raft node exists. Rerun on images from
0c2a1801: converged 17 s after the Tombstone, all 12 leaders logged `snapshot
after create raft node`, 0 of 1494 writes lost
([evidence](https://github.com/bitflicker64/hugegraph/tree/logs/fix-3227-review)).
##########
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();
+ }
+ }
+
+ List<Long> peerIds = new ArrayList<>();
+ for (String peer : RaftUtils.getPeerEndpoints(raftNode)) {
+ Long id = pdIds.getOrDefault(peer.toLowerCase(),
localIds.get(peer.toLowerCase()));
+ if (id != null) {
+ peerIds.add(id);
+ }
+ }
+ List<Long> learners = new ArrayList<>();
+ for (String learner : RaftUtils.getLearnerEndpoints(raftNode)) {
+ Long id = pdIds.getOrDefault(learner.toLowerCase(),
+ localIds.get(learner.toLowerCase()));
+ if (id != null) {
+ learners.add(id);
+ }
+ }
+ shardGroup.changeShardList(peerIds, learners,
partitionManager.getStore().getId());
+ partitionManager.updateShardGroup(shardGroup);
+ try {
+
partitionManager.getPdProvider().updateShardGroup(shardGroup.getProtoObj());
+ } catch (PDException e) {
+ log.warn("Raft {} update shard group to pd failed, {}",
getGroupId(), e.getMessage());
+ }
+ log.info("Raft {} shard group after store id change {}", getGroupId(),
+ shardGroup.getMetaPbShard());
+
+ return waitForReplicate(changed) ? HgRaftError.OK.toStatus() :
+ HgRaftError.TASK_CONTINUE.toStatus();
Review Comment:
97f5b013: a timeout now logs a warning and returns `TASK_ERROR`, so it no
longer re-queues a pass that finds nothing to do and reports OK. The ids stay
taken before the wait on purpose: a rebuilt Store that restarts before PD names
it exits in `PartitionManager.loadPartitions` (`System.exit(0)` at line 302),
so moving them after the wait would crash-loop it during catch-up. Nothing is
lost on a timeout: the node exists and its replicator keeps installing the
snapshot on its own; the task result only goes to the log. The timeout path
needs a live raft group, so it has no unit test; the kind rerun in the PR
description covers the normal path.
##########
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) {
Review Comment:
0c2a1801 adds the procedure to `hugegraph-store/docs/operations-guide.md`
(Single Store Node Failure, step 3): retire the old id with `Tombstone`, run
`patrolPartitions`, verify, then delete it, with the note that Store versions
without this fix never finish it. The website only lists these PD endpoints
(quickstart hugegraph-pd), with no replacement procedure to correct. The Helm
chart README is not in master yet; it lives in #3218, and its "do not delete a
Store volume" warning changes there with a version caveat once this is merged
and rerun on master images.
--
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]
