diff --git a/hugegraph-store/docs/operations-guide.md b/hugegraph-store/docs/operations-guide.md index 78e1c718b5..17f3fe3630 100644 --- a/hugegraph-store/docs/operations-guide.md +++ b/hugegraph-store/docs/operations-guide.md @@ -654,6 +654,17 @@ curl http://192.168.1.10:8620/v1/partitionsAndStatus 5. **Verify**: Wait for even distribution (may take hours) +Replica migration can leave learners in a Raft group while they receive a snapshot or catch up. +If the same membership change is retried after a catch-up timeout, Store resumes those learners +and promotes them to voting members after catch-up; it does not need to create them again. +Each catch-up check waits up to 600 seconds before returning `TASK_CONTINUE`. This is a +per-attempt wait, not an overall migration deadline; a peer that cannot catch up can keep +the operation pending across retries. Inspect its replication progress and errors before +continuing a node operation. This retry correction does not add durable task admission +or recovery of a migration task across a Leader change. +Verify the actual voter and learner lists before treating migration as complete: a target count +that includes learners is not proof of voting membership. + ### Removing Store Nodes **When to Remove**: diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java index 85e43fea02..7c65e08ea1 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java @@ -295,23 +295,26 @@ public ShardGroup getShardGroup() { * @return true means completed, false means not completed */ public Status changePeers(List peers, final Closure done) { - if (ListUtils.isEqualList(peers, RaftUtils.getPeerEndpoints(raftNode))) { + Configuration oldConf = getCurrentConf(); + List oldVoters = RaftUtils.getPeerEndpoints(oldConf); + if (ListUtils.isEqualList(peers, oldVoters) && oldConf.listLearners().isEmpty()) { return Status.OK(); } Status result = HgRaftError.TASK_CONTINUE.toStatus(); - List oldPeers = RaftUtils.getAllEndpoints(raftNode); + List oldPeers = RaftUtils.getAllEndpoints(oldConf); log.info("Raft {} changePeers start, old peer is {}, new peer is {}", getGroupId(), oldPeers, peers); // Check the peer that needs to be added. List addPeers = ListUtils.removeAll(peers, oldPeers); + // A retry must still promote target peers that already exist as learners. + List pendingPromotions = ListUtils.removeAll(peers, oldVoters); // learner to be deleted. Possible peer change. List removedPeers = ListUtils.removeAll(oldPeers, peers); HgCmdClient rpcClient = storeEngine.getHgCmdClient(); // Generate a new Configuration object - Configuration oldConf = getCurrentConf(); Configuration conf = oldConf.copy(); FutureClosure closure; @@ -341,9 +344,11 @@ public Status changePeers(List peers, final Closure done) { status.getErrorMsg()); return status; } + } - // 2.2 Waiting learner to synchronize snapshot (check added learner) - if (!waitForReplicate(addPeers)) { + if (!pendingPromotions.isEmpty()) { + // 2.2 Wait for both newly added and previously created target learners. + if (!waitForReplicate(pendingPromotions)) { return HgRaftError.TASK_CONTINUE.toStatus(); } @@ -351,13 +356,16 @@ public Status changePeers(List peers, final Closure done) { closure = new FutureClosure(); // 2.3 change learner to follower (first remove, then add follower) - raftNode.removeLearners(conf.listLearners(), closure); - if (!closure.get().isOk()) { + List promoting = new ArrayList<>(); + pendingPromotions.forEach(peer -> promoting.add(JRaftUtils.getPeerId(peer))); + raftNode.removeLearners(promoting, closure); + var status = closure.get(); + if (!status.isOk()) { log.error("Raft {} remove learner error, result:{}", getGroupId(), status); return HgRaftError.TASK_ERROR.toStatus(); } - addPeers.forEach(peer -> { + pendingPromotions.forEach(peer -> { conf.removeLearner(JRaftUtils.getPeerId(peer)); conf.addPeer(JRaftUtils.getPeerId(peer)); }); @@ -365,7 +373,8 @@ public Status changePeers(List peers, final Closure done) { // add follower closure = new FutureClosure(); raftNode.changePeers(conf, closure); - if (!closure.get().isOk()) { + status = closure.get(); + if (!status.isOk()) { log.error("Raft {} changePeers error, result:{}", getGroupId(), status); return HgRaftError.TASK_ERROR.toStatus(); } @@ -419,7 +428,7 @@ public Status changePeers(List peers, final Closure done) { * * @return false if the peers did not catch up within 600s */ - private boolean waitForReplicate(List peers) { + boolean waitForReplicate(List peers) { //todo Each learner will wait for 1s, if another one is not sync.Consider using // countdownLatch boolean allLearnerSnapshotOk = false; diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index 28ca7102bf..055f884424 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -246,6 +246,7 @@ **/IpUtilTest.java **/PartitionStateMachineTest.java **/PartitionEngineErrorTest.java + **/PartitionEngineMembershipTest.java **/BatchGraphIsolationTest.java **/GraphIDManagerTest.java **/SchemaDriverTest.java diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionEngineMembershipTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionEngineMembershipTest.java new file mode 100644 index 0000000000..966cae1ec8 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionEngineMembershipTest.java @@ -0,0 +1,332 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hugegraph.store; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hugegraph.store.cmd.HgCmdClient; +import org.apache.hugegraph.store.meta.PartitionManager; +import org.apache.hugegraph.store.meta.ShardGroup; +import org.apache.hugegraph.store.raft.util.RaftUtils; +import org.apache.hugegraph.store.util.HgRaftError; +import org.junit.Test; + +import com.alipay.sofa.jraft.Closure; +import com.alipay.sofa.jraft.JRaftUtils; +import com.alipay.sofa.jraft.Node; +import com.alipay.sofa.jraft.Status; +import com.alipay.sofa.jraft.conf.Configuration; +import com.alipay.sofa.jraft.entity.NodeId; +import com.alipay.sofa.jraft.entity.PeerId; +import com.alipay.sofa.jraft.error.RaftError; + +public class PartitionEngineMembershipTest { + + private static final String A = "127.0.0.1:18510"; + private static final String B = "127.0.0.1:18511"; + private static final String C = "127.0.0.1:18512"; + private static final String D = "127.0.0.1:18513"; + private static final List TARGET = List.of(A, B, C); + + @Test + public void testResumesExistingLearnersWithoutRecreatingNodes() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of(B, C)); + + assertTrue(fixture.engine.changePeers(TARGET, null).isOk()); + + fixture.assertMembership(TARGET, List.of()); + assertEquals(List.of(List.of(B, C)), fixture.engine.waitedPeers); + assertTrue(fixture.created.isEmpty()); + assertEquals(0, fixture.engine.snapshots); + } + + @Test + public void testPromotesExistingAndNewLearnersTogether() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of(B)); + + assertTrue(fixture.engine.changePeers(TARGET, null).isOk()); + + fixture.assertMembership(TARGET, List.of()); + assertEquals(List.of(C), fixture.created); + assertEquals(List.of(List.of(B, C)), fixture.engine.waitedPeers); + } + + @Test + public void testResumesAfterOneLearnerWasAlreadyPromoted() throws Exception { + Fixture fixture = new Fixture(List.of(A, B), List.of(C)); + + assertTrue(fixture.engine.changePeers(TARGET, null).isOk()); + + fixture.assertMembership(TARGET, List.of()); + assertEquals(List.of(List.of(C)), fixture.engine.waitedPeers); + assertTrue(fixture.created.isEmpty()); + } + + @Test + public void testCatchupTimeoutResumesWithoutRecreatingLearners() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of()); + fixture.engine.waitResults.add(false); + fixture.engine.waitResults.add(true); + + Status first = fixture.engine.changePeers(TARGET, null); + assertEquals(HgRaftError.TASK_CONTINUE.getNumber(), first.getCode()); + fixture.assertMembership(List.of(A), List.of(B, C)); + + assertTrue(fixture.engine.changePeers(TARGET, null).isOk()); + fixture.assertMembership(TARGET, List.of()); + assertEquals(List.of(B, C), fixture.created); + assertEquals(List.of(List.of(B, C), List.of(B, C)), fixture.engine.waitedPeers); + } + + @Test + public void testExistingLearnersRemainLearnersUntilCaughtUp() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of(B, C)); + fixture.engine.waitResults.add(false); + + Status result = fixture.engine.changePeers(TARGET, null); + + assertEquals(HgRaftError.TASK_CONTINUE.getNumber(), result.getCode()); + fixture.assertMembership(List.of(A), List.of(B, C)); + verify(fixture.node, never()).removeLearners(anyList(), any(Closure.class)); + assertTrue(fixture.changed.isEmpty()); + } + + @Test + public void testLearnerRemovalFailureStopsPromotion() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of(B, C)); + fixture.removeResults.add(new Status(RaftError.EIO, "cannot remove learners")); + + Status result = fixture.engine.changePeers(TARGET, null); + + assertFalse(result.isOk()); + fixture.assertMembership(List.of(A), List.of(B, C)); + assertTrue(fixture.changed.isEmpty()); + assertTrue(fixture.destroyed.isEmpty()); + } + + @Test + public void testPromotionFailureIsNotReportedAsSuccess() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of(B, C)); + fixture.changeResults.add(new Status(RaftError.EBUSY, "configuration busy")); + + Status result = fixture.engine.changePeers(TARGET, null); + + assertFalse(result.isOk()); + fixture.assertMembership(List.of(A), List.of()); + assertTrue(fixture.destroyed.isEmpty()); + } + + @Test + public void testRetriesAfterLearnersWereRemovedButPromotionFailed() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of(B, C)); + fixture.changeResults.add(new Status(RaftError.EBUSY, "configuration busy")); + + Status first = fixture.engine.changePeers(TARGET, null); + + assertEquals(HgRaftError.TASK_ERROR.getNumber(), first.getCode()); + fixture.assertMembership(List.of(A), List.of()); + assertTrue(fixture.created.isEmpty()); + + assertTrue(fixture.engine.changePeers(TARGET, null).isOk()); + fixture.assertMembership(TARGET, List.of()); + assertEquals(List.of(B, C), fixture.created); + assertEquals(List.of(List.of(B, C), List.of(B, C)), fixture.engine.waitedPeers); + assertTrue(fixture.destroyed.isEmpty()); + } + + @Test + public void testRemovesStaleLearnerEvenWhenVotersAlreadyMatch() throws Exception { + Fixture fixture = new Fixture(TARGET, List.of(D)); + + assertTrue(fixture.engine.changePeers(TARGET, null).isOk()); + + fixture.assertMembership(TARGET, List.of()); + assertEquals(List.of(D), fixture.destroyed); + assertTrue(fixture.engine.waitedPeers.isEmpty()); + } + + @Test + public void testKeepsStaleLearnerIfTargetPromotionFails() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of(B, C, D)); + fixture.changeResults.add(new Status(RaftError.EBUSY, "configuration busy")); + + Status result = fixture.engine.changePeers(TARGET, null); + + assertFalse(result.isOk()); + fixture.assertMembership(List.of(A), List.of(D)); + assertTrue(fixture.destroyed.isEmpty()); + } + + @Test + public void testHealthyTargetRemainsANoOp() throws Exception { + Fixture fixture = new Fixture(TARGET, List.of()); + + assertTrue(fixture.engine.changePeers(TARGET, null).isOk()); + + assertTrue(fixture.changed.isEmpty()); + assertTrue(fixture.created.isEmpty()); + assertTrue(fixture.destroyed.isEmpty()); + assertTrue(fixture.engine.waitedPeers.isEmpty()); + } + + @Test + public void testFailedLearnerConfigurationDoesNotStartPromotion() throws Exception { + Fixture fixture = new Fixture(List.of(A), List.of()); + fixture.changeResults.add(new Status(RaftError.EIO, "cannot add learners")); + + Status result = fixture.engine.changePeers(TARGET, null); + + assertFalse(result.isOk()); + fixture.assertMembership(List.of(A), List.of()); + assertTrue(fixture.engine.waitedPeers.isEmpty()); + verify(fixture.node, never()).removeLearners(anyList(), any(Closure.class)); + } + + @Test + public void testRemovingLeaderStillTransfersAndRequestsContinuation() throws Exception { + Fixture fixture = new Fixture(TARGET, List.of(D)); + + Status result = fixture.engine.changePeers(List.of(B, C, D), null); + + assertEquals(HgRaftError.TASK_CONTINUE.getNumber(), result.getCode()); + fixture.assertMembership(List.of(A, B, C, D), List.of()); + verify(fixture.node).transferLeadershipTo(PeerId.ANY_PEER); + assertTrue(fixture.destroyed.isEmpty()); + } + + private static class Fixture { + + private final Node node = mock(Node.class); + private final TestEngine engine; + private final AtomicReference membership; + private final Queue changeResults = new ArrayDeque<>(); + private final Queue removeResults = new ArrayDeque<>(); + private final List changed = new ArrayList<>(); + private final List created = new ArrayList<>(); + private final List destroyed = new ArrayList<>(); + + Fixture(List voters, List learners) throws Exception { + Configuration initial = new Configuration(); + voters.forEach(peer -> initial.addPeer(JRaftUtils.getPeerId(peer))); + learners.forEach(peer -> initial.addLearner(JRaftUtils.getPeerId(peer))); + this.membership = new AtomicReference<>(initial); + HgStoreEngine store = mock(HgStoreEngine.class); + PartitionManager manager = mock(PartitionManager.class); + HgCmdClient client = mock(HgCmdClient.class); + when(store.getPartitionManager()).thenReturn(manager); + when(store.getHgCmdClient()).thenReturn(client); + this.engine = new TestEngine(store); + Field raftNode = PartitionEngine.class.getDeclaredField("raftNode"); + raftNode.setAccessible(true); + raftNode.set(this.engine, this.node); + when(this.node.listPeers()).thenAnswer(invocation -> + new ArrayList<>(this.membership.get().listPeers())); + when(this.node.listLearners()).thenAnswer(invocation -> + new ArrayList<>(this.membership.get().listLearners())); + when(this.node.getNodeId()).thenReturn(new NodeId("hg_17", JRaftUtils.getPeerId(A))); + doAnswer(invocation -> { + Configuration next = ((Configuration) invocation.getArgument(0)).copy(); + this.changed.add(next); + Status status = this.changeResults.isEmpty() ? Status.OK() : this.changeResults.remove(); + if (status.isOk()) { + this.membership.set(next.copy()); + } + ((Closure) invocation.getArgument(1)).run(status); + return null; + }).when(this.node).changePeers(any(Configuration.class), any(Closure.class)); + doAnswer(invocation -> { + List removing = invocation.getArgument(0); + Status status = this.removeResults.isEmpty() ? Status.OK() : this.removeResults.remove(); + if (status.isOk()) { + Configuration next = this.membership.get().copy(); + removing.forEach(next::removeLearner); + this.membership.set(next); + } + ((Closure) invocation.getArgument(1)).run(status); + return null; + }).when(this.node).removeLearners(anyList(), any(Closure.class)); + doAnswer(invocation -> { + this.created.add(invocation.getArgument(0)); + ((Closure) invocation.getArgument(3)).run(Status.OK()); + return CompletableFuture.completedFuture(null); + }).when(client).createRaftNode(anyString(), anyList(), any(Configuration.class), any(Closure.class)); + doAnswer(invocation -> { + String peer = invocation.getArgument(0); + assertFalse("destroy only after the peer is removed from the committed configuration", + RaftUtils.getAllEndpoints(this.membership.get()).contains(peer)); + this.destroyed.add(peer); + ((Closure) invocation.getArgument(2)).run(Status.OK()); + return CompletableFuture.completedFuture(null); + }).when(client).destroyRaftNode(anyString(), anyList(), any(Closure.class)); + } + + void assertMembership(List voters, List learners) { + assertEquals(new HashSet<>(voters), + new HashSet<>(RaftUtils.getPeerEndpoints(this.membership.get()))); + assertEquals(new HashSet<>(learners), + new HashSet<>(RaftUtils.getLearnerEndpoints(this.membership.get()))); + } + } + + private static class TestEngine extends PartitionEngine { + + private final Queue waitResults = new ArrayDeque<>(); + private final List> waitedPeers = new ArrayList<>(); + private int snapshots; + + TestEngine(HgStoreEngine store) { + super(store, mock(ShardGroup.class)); + } + + @Override + public Integer getGroupId() { + return 17; + } + + @Override + boolean waitForReplicate(List peers) { + this.waitedPeers.add(new ArrayList<>(peers)); + return this.waitResults.isEmpty() || this.waitResults.remove(); + } + + @Override + protected void doSnapshot(Closure done) { + this.snapshots++; + done.run(Status.OK()); + } + } +}