diff --git a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/PDClient.java b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/PDClient.java index 1b0ce4982e..3857443d86 100644 --- a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/PDClient.java +++ b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/PDClient.java @@ -621,17 +621,7 @@ public KVPair getPartitionById(String graphName, KVPair partShard = cache.getPartitionById(graphName, partId); if (partShard == null) { - Pdpb.GetPartitionByIDRequest request = Pdpb.GetPartitionByIDRequest.newBuilder() - .setHeader(header) - .setGraphName( - graphName) - .setPartitionId( - partId) - .build(); - GetPartitionResponse response = - blockingUnaryCall(PDGrpc.getGetPartitionByIDMethod(), request); - handleResponseError(response.getHeader()); - partShard = new KVPair<>(response.getPartition(), response.getLeader()); + partShard = getPartitionByIdDirect(graphName, partId); if (config.isEnableCache()) { cache.update(graphName, partShard.getKey().getId(), partShard.getKey()); cache.updateShardGroup(getShardGroup(partShard.getKey().getId())); @@ -653,6 +643,19 @@ public KVPair getPartitionById(String graphName, return partShard; } + /** + * Query this graph partition without the cache or any additional shard-group lookup. + * In particular, a NOT_FOUND error belongs to this partition RPC rather than cache filling. + */ + public KVPair getPartitionByIdDirect(String graphName, + int partId) throws PDException { + Pdpb.GetPartitionByIDRequest request = Pdpb.GetPartitionByIDRequest.newBuilder() + .setHeader(header).setGraphName(graphName).setPartitionId(partId).build(); + GetPartitionResponse response = blockingUnaryCall(PDGrpc.getGetPartitionByIDMethod(), request); + handleResponseError(response.getHeader()); + return new KVPair<>(response.getPartition(), response.getLeader()); + } + public ShardGroup getShardGroup(int partId) throws PDException { ShardGroup group = cache.getShardGroup(partId); if (group == null) { diff --git a/hugegraph-store/docs/operations-guide.md b/hugegraph-store/docs/operations-guide.md index 17f3fe3630..aafd8fa424 100644 --- a/hugegraph-store/docs/operations-guide.md +++ b/hugegraph-store/docs/operations-guide.md @@ -665,6 +665,27 @@ 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. +### Restoring Store Membership after a Restart + +For a persisted partition, startup verifies the shard group directly with PD and refreshes +the derived local `SHARDGROUP` once per group when this Store is still a member and no +engine is running for the group. This lets JRaft restore its durable configuration even +when the business database contains older membership metadata. Running engines and +durable Raft configuration are not replaced by the PD bootstrap metadata. + +Only PD's explicit `NOT_FOUND` response for a persisted graph partition skips that graph's +recovery, for example when it was dropped while the Store was offline. Other graphs can +still recover. The skipped graph's local data, partition-store location and Raft files +remain intact; a group with no remaining graph partitions does not start an engine. +A null/ambiguous result, invalid metadata or a failed PD lookup blocks recovery of the +whole Store process while retaining local data. This includes members with an unspecified +`None` role until PD reports usable member roles. Restore PD access or correct the reported +metadata before retrying startup; do not delete data or reset Raft peers to bypass a failure. +PD exclusion is not proof that a Store has been safely retired. Check every assigned +group after a restart before continuing a rolling operation. This recovery correction +does not establish why older derived metadata was persisted, or provide a complete +desired/observed membership protocol or protection against partial PD partition queries. + ### Removing Store Nodes **When to Remove**: diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java index 9325ff71d9..85a281e940 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java @@ -36,6 +36,7 @@ import org.apache.hugegraph.pd.common.PDException; import org.apache.hugegraph.pd.grpc.MetaTask; import org.apache.hugegraph.pd.grpc.Metapb; +import org.apache.hugegraph.pd.grpc.Pdpb; import org.apache.hugegraph.store.HgStoreEngine; import org.apache.hugegraph.store.business.BusinessHandlerImpl; import org.apache.hugegraph.store.cmd.HgCmdClient; @@ -107,6 +108,10 @@ public static final class InvalidShardException extends IllegalStateException { private InvalidShardException() { super("Local partition does not belong to this store"); } + + private InvalidShardException(String message, Throwable cause) { + super(message, cause); + } } public void loadPartition() { @@ -259,6 +264,7 @@ private void loadPartitions() { } Set normalPartitions = new HashSet<>(); + Set retainedPartitions = new HashSet<>(); // Once according to the partition read for (int partId : partIds) { @@ -269,64 +275,108 @@ private void loadPartitions() { } var metaParts = wrapper.scan(partId, Metapb.Partition.parser(), key); - int countOfPartition = 0; - - var shards = pdProvider.getShardGroup(partId).getShardsList(); + if (metaParts.isEmpty()) { + wrapper.close(partId); + continue; + } + // A migrated replica can have an older derived SHARDGROUP in its data DB than + // its durable Raft configuration. Validate recovery against PD before Raft replay. + var pdGroup = recoveryShardGroup(partId); + var shards = pdGroup.getShardsList(); + if (shards.stream().noneMatch(s -> s.getStoreId() == storeId)) { + log.error("partition {} is illegal. store id {} not in valid shard group:{}", + partId, storeId, shards2Peers(shards)); + throw new InvalidShardException(); + } + List recoverableParts = new ArrayList<>(); for (var metaPart : metaParts) { var graph = metaPart.getGraphName(); - var pdPartition = pdProvider.getPartitionByID(graph, metaPart.getId()); - boolean isLegeal = false; - - if (pdPartition != null) { - // Check if it contains this store id - if (shards.stream().anyMatch(s -> s.getStoreId() == storeId)) { - isLegeal = true; - } - } else { - continue; + if (metaPart.getId() != partId || graph.isEmpty()) { + throw new InvalidShardException("Invalid persisted partition " + graph + + "-" + partId, null); } - - if (isLegeal) { - if (!partitions.containsKey(graph)) { - partitions.put(graph, new ConcurrentHashMap<>()); + Partition pdPartition; + try { + pdPartition = pdProvider.getPartitionByIDForRecovery(graph, metaPart.getId()); + } catch (PDException e) { + if (e.getErrorCode() == Pdpb.ErrorType.NOT_FOUND_VALUE) { + // A graph can be dropped while this Store is offline. Do not serve it or + // infer permission to delete its data, Store path or Raft metadata. + retainedPartitions.add(partId); + log.warn("PD no longer has partition {}-{}; skipping recovery and " + + "retaining local data", graph, partId); + continue; } - - countOfPartition += 1; - - Partition partition = new Partition(metaPart); - partition.setWorkState( - Metapb.PartitionState.PState_Normal); // Start recovery work state - partitions.get(graph).put(partition.getId(), partition); - log.info("load partition : {} -{}", partition.getGraphName(), - partition.getId()); - } else { - // Invalid - // removePartitionFromLocalDb(graph, partId); - // var businessHandler = HgStoreEngine.getInstance().getBusinessHandler(); - // businessHandler.truncate(graph, partId); - // businessHandler.dbCompaction(graph, partId); - log.error("partition {}-{} is illegal. store id {} not in valid shard group:{}", - graph, partId, getStore().getId(), shards2Peers(shards)); - throw new InvalidShardException(); + throw new InvalidShardException("Cannot read persisted partition " + graph + + "-" + partId + " from PD", e); + } catch (RuntimeException e) { + throw new InvalidShardException("Cannot read persisted partition " + graph + + "-" + partId + " from PD", e); } + if (pdPartition == null || pdPartition.getId() != partId || + !graph.equals(pdPartition.getGraphName())) { + throw new InvalidShardException("Cannot verify persisted partition " + graph + + "-" + partId + " with PD", null); + } + recoverableParts.add(metaPart); + } + if (recoverableParts.isEmpty()) { + wrapper.close(partId); + continue; } - if (countOfPartition > 0) { - // Partition data is normal - normalPartitions.add(partId); + // Only refresh startup metadata, once per group even when several graphs share it. + // The running engine and JRaft's durable configuration remain authoritative. + if (HgStoreEngine.getInstance().getPartitionEngine(partId) == null) { + updateShardGroup(ShardGroup.from(pdGroup)); } + for (var metaPart : recoverableParts) { + var graph = metaPart.getGraphName(); + partitions.computeIfAbsent(graph, g -> new ConcurrentHashMap<>()); + Partition partition = new Partition(metaPart); + partition.setWorkState(Metapb.PartitionState.PState_Normal); + partitions.get(graph).put(partition.getId(), partition); + log.info("load partition : {} -{}", partition.getGraphName(), partition.getId()); + } + normalPartitions.add(partId); wrapper.close(partId); } // Remove redundant partition storage paths, partitions that have been migrated away may migrate back for (var location : storeMetadata.getPartitionStores()) { - if (!normalPartitions.contains(location.getPartitionId())) { + if (!normalPartitions.contains(location.getPartitionId()) && + !retainedPartitions.contains(location.getPartitionId())) { storeMetadata.removePartitionStore(location.getPartitionId()); } } } + private Metapb.ShardGroup recoveryShardGroup(int partId) { + Metapb.ShardGroup group; + try { + group = pdProvider.getShardGroupDirect(partId); + } catch (RuntimeException e) { + throw new InvalidShardException("Cannot read recovery shard group " + partId + + " from PD", e); + } + if (group == null || group.getId() != partId || group.getShardsCount() == 0) { + throw new InvalidShardException("Invalid recovery shard group " + partId + + " from PD", null); + } + Set stores = new HashSet<>(); + for (var shard : group.getShardsList()) { + if (shard.getStoreId() == 0 || !stores.add(shard.getStoreId()) || + (shard.getRole() != Metapb.ShardRole.Leader && + shard.getRole() != Metapb.ShardRole.Follower && + shard.getRole() != Metapb.ShardRole.Learner)) { + throw new InvalidShardException("Invalid recovery shard members for group " + + partId + " from PD", null); + } + } + return group; + } + public List loadPartitionsFromDb(int partitionId) { byte[] key = MetadataKeyHelper.getPartitionPrefixKey(); return wrapper.scan(partitionId, Metapb.Partition.parser(), key); @@ -797,7 +847,11 @@ public List getPartitionPeers(ShardGroup shardGroup) { for (Shard shard : shardGroup.getShards()) { Store store = getStore(shard.getStoreId()); if (store != null && !store.getRaftAddress().isEmpty()) { - peers.add(store.getRaftAddress() + "::" + priority); + String peer = store.getRaftAddress() + "::" + priority; + if (shard.getRole() == Metapb.ShardRole.Learner) { + peer += "/learner"; + } + peers.add(peer); final int gap = Math.max(decayPriorityGap, (priority / 5)); priority = Math.max(ElectionPriority.MinValue, (priority - gap)); } diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/DefaultPdProvider.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/DefaultPdProvider.java index 1a99f27feb..0d1035be55 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/DefaultPdProvider.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/DefaultPdProvider.java @@ -173,6 +173,12 @@ public Partition getPartitionByID(String graph, int partId) { return null; } + @Override + public Partition getPartitionByIDForRecovery(String graph, int partId) throws PDException { + KVPair pair = pdClient.getPartitionByIdDirect(graph, partId); + return pair == null ? null : new Partition(pair.getKey()); + } + @Override public Metapb.Shard getPartitionLeader(String graph, int partId) { try { diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/PdProvider.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/PdProvider.java index 7d028965c4..090b40024b 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/PdProvider.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/pd/PdProvider.java @@ -43,6 +43,14 @@ public interface PdProvider { Partition getPartitionByID(String graph, int partId); + /** + * Preserve PD lookup failures for recovery. Only an explicit NOT_FOUND error proves absence; + * the legacy lookup's null result remains ambiguous and must not authorize cleanup. + */ + default Partition getPartitionByIDForRecovery(String graph, int partId) throws PDException { + return getPartitionByID(graph, partId); + } + Metapb.Shard getPartitionLeader(String graph, int partId); Metapb.Partition getPartitionByCode(String graph, int code); diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index 055f884424..f15b8f9ed2 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -247,6 +247,8 @@ **/PartitionStateMachineTest.java **/PartitionEngineErrorTest.java **/PartitionEngineMembershipTest.java + **/PartitionRecoveryMembershipTest.java + **/RecoveryPdLookupTest.java **/BatchGraphIsolationTest.java **/GraphIDManagerTest.java **/SchemaDriverTest.java diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/HeartbeatShutdownTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/HeartbeatShutdownTest.java index 7a89ddf241..b401493519 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/HeartbeatShutdownTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/HeartbeatShutdownTest.java @@ -192,10 +192,10 @@ private static void runInvalidShardRecovery(Path marker) throws Exception { options.setRaftPath(raft.toString()); PdProvider provider = Mockito.mock(PdProvider.class); Metapb.Partition partition = Metapb.Partition.newBuilder().setId(1).setGraphName("graph").build(); - Mockito.when(provider.getShardGroup(1)).thenReturn(Metapb.ShardGroup.newBuilder() + Mockito.when(provider.getShardGroupDirect(1)).thenReturn(Metapb.ShardGroup.newBuilder() .setId(1).addShards(Metapb.Shard.newBuilder().setStoreId(2) .setRole(Metapb.ShardRole.Leader)).build()); - Mockito.when(provider.getPartitionByID("graph", 1)).thenReturn(new Partition(partition)); + Mockito.when(provider.getPartitionByIDForRecovery("graph", 1)).thenReturn(new Partition(partition)); Store remote = new Store(); remote.setRaftAddress("127.0.0.1:8511"); Mockito.when(provider.getStoreByID(2L)).thenReturn(remote); diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionRecoveryMembershipTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionRecoveryMembershipTest.java new file mode 100644 index 0000000000..4c113ca647 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionRecoveryMembershipTest.java @@ -0,0 +1,483 @@ +/* + * 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.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hugegraph.pd.client.PDClient; +import org.apache.hugegraph.pd.common.KVPair; +import org.apache.hugegraph.pd.common.PDException; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.apache.hugegraph.pd.grpc.Pdpb; +import org.apache.hugegraph.store.business.BusinessHandler; +import org.apache.hugegraph.store.meta.GraphManager; +import org.apache.hugegraph.store.meta.MetadataKeyHelper; +import org.apache.hugegraph.store.meta.Partition; +import org.apache.hugegraph.store.meta.PartitionManager; +import org.apache.hugegraph.store.meta.ShardGroup; +import org.apache.hugegraph.store.meta.Store; +import org.apache.hugegraph.store.meta.StoreMetadata; +import org.apache.hugegraph.store.options.HgStoreEngineOptions; +import org.apache.hugegraph.store.pd.DefaultPdProvider; +import org.apache.hugegraph.store.pd.PdProvider; +import org.apache.hugegraph.store.util.PartitionMetaStoreWrapper; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import com.alipay.sofa.jraft.JRaftUtils; +import com.alipay.sofa.jraft.conf.Configuration; +import com.alipay.sofa.jraft.conf.ConfigurationManager; +import com.alipay.sofa.jraft.core.NodeImpl; +import com.alipay.sofa.jraft.entity.EnumOutter.EntryType; +import com.alipay.sofa.jraft.entity.LogEntry; +import com.alipay.sofa.jraft.entity.LogId; +import com.alipay.sofa.jraft.entity.codec.v2.LogEntryV2CodecFactory; +import com.alipay.sofa.jraft.option.LogStorageOptions; +import com.alipay.sofa.jraft.option.RaftOptions; +import com.alipay.sofa.jraft.rpc.RpcServer; +import com.alipay.sofa.jraft.storage.LogStorage; +import com.alipay.sofa.jraft.storage.impl.RocksDBLogStorage; + +public class PartitionRecoveryMembershipTest { + + private static final int GROUP = 0; + private static final long SELF = 3L; + private static final String SELF_ADDRESS = "127.0.0.1:18539"; + private static final String OTHER_ADDRESS = "127.0.0.1:18538"; + + @Rule + public TemporaryFolder folder = new TemporaryFolder(); + + private final HgStoreEngine engine = HgStoreEngine.getInstance(); + private final Map originalFields = new HashMap<>(); + private PdProvider pd; + private PartitionManager manager; + private PartitionMetaStoreWrapper wrapper; + private StoreMetadata metadata; + private HgStoreEngineOptions options; + private List persisted; + private AtomicReference localGroup; + private Metapb.ShardGroup directGroup; + private Path data; + private Map engines; + private HeartbeatService heartbeat; + private CountDownLatch leaderPublication; + + @Before + @SuppressWarnings("unchecked") + public void setUp() throws Exception { + data = folder.newFolder().toPath(); + Files.createDirectories(data.resolve("db/00000")); + Files.createDirectories(data.resolve("raft/00000/log")); + options = new HgStoreEngineOptions(); + options.setDataPath(data.toString()); + options.setRaftPath(data.toString()); + options.setRaftAddress(SELF_ADDRESS); + options.getRaftOptions().setUseRocksDBSegmentLogStorage(false); + options.getRaftOptions().setSnapshotIntervalSecs(0); + pd = mock(PdProvider.class); + manager = new PartitionManager(pd, options); + wrapper = mock(PartitionMetaStoreWrapper.class); + metadata = mock(StoreMetadata.class); + setField(manager, "wrapper", wrapper); + setField(manager, "storeMetadata", metadata); + setField(manager, "graphManager", mock(GraphManager.class)); + persisted = new ArrayList<>(List.of(partition("graph"))); + localGroup = new AtomicReference<>(group(1L)); + directGroup = group(SELF, 1L); + Store self = store(SELF, SELF_ADDRESS); + when(metadata.getStore()).thenReturn(self); + when(metadata.getPartitionStore(GROUP)).thenReturn(Metapb.PartitionStore.newBuilder() + .setPartitionId(GROUP).setStoreLocation(data.toString()).build()); + when(metadata.getPartitionRaft(GROUP)).thenReturn(Metapb.PartitionRaft.newBuilder() + .setPartitionId(GROUP).setRaftLocation(data.toString()).build()); + when(metadata.getPartitionRaftLocation(GROUP)).thenReturn(data.toString()); + when(metadata.getPartitionStoreLocation(eq(GROUP), anyString())).thenReturn(data.toString()); + when(metadata.getPartitionStores()).thenAnswer(call -> List.of(metadata.getPartitionStore(GROUP))); + when(wrapper.scan(eq(GROUP), eq(Metapb.Partition.parser()), any(byte[].class))) + .thenAnswer(call -> new ArrayList<>(persisted)); + when(wrapper.get(eq(GROUP), any(byte[].class), eq(Metapb.ShardGroup.parser()))) + .thenAnswer(call -> localGroup.get()); + doAnswer(call -> { + byte[] key = call.getArgument(1); + if (Arrays.equals(key, MetadataKeyHelper.getShardGroupKey(GROUP))) { + localGroup.set(Metapb.ShardGroup.parseFrom((byte[]) call.getArgument(2))); + } + return null; + }).when(wrapper).put(eq(GROUP), any(byte[].class), any(byte[].class)); + when(pd.getShardGroupDirect(GROUP)).thenAnswer(call -> directGroup); + when(pd.getShardGroup(GROUP)).thenAnswer(call -> directGroup); + leaderPublication = new CountDownLatch(1); + doAnswer(call -> { + for (StackTraceElement frame : Thread.currentThread().getStackTrace()) { + if (frame.getClassName().equals(PartitionEngine.class.getName()) && + frame.getMethodName().equals("onLeaderStart")) { + leaderPublication.countDown(); + } + } + return null; + }).when(pd).updateShardGroup(any(Metapb.ShardGroup.class)); + when(pd.getPartitionByIDForRecovery(anyString(), eq(GROUP))) + .thenAnswer(call -> new Partition(partition(call.getArgument(0)))); + when(pd.getPartitionsByStore(SELF)).thenAnswer(call -> { + List parts = new ArrayList<>(); + persisted.forEach(p -> parts.add(new Partition(p))); + return parts; + }); + when(pd.getStoreByID(anyLong())).thenAnswer(call -> { + long id = call.getArgument(0); + return id == SELF ? self : store(id, OTHER_ADDRESS); + }); + Field map = HgStoreEngine.class.getDeclaredField("partitionEngines"); + map.setAccessible(true); + engines = (Map) map.get(engine); + assertTrue("Recovery tests require an isolated Store", engines.isEmpty()); + replaceEngineField("partitionManager", manager); + replaceEngineField("options", options); + replaceEngineField("pdProvider", pd); + replaceEngineField("businessHandler", mock(BusinessHandler.class)); + heartbeat = mock(HeartbeatService.class); + replaceEngineField("heartbeatService", heartbeat); + RpcServer rpc = mock(RpcServer.class); + when(rpc.boundPort()).thenReturn(18539); + replaceEngineField("rpcServer", rpc); + } + + @After + public void tearDown() throws Exception { + if (engines == null) { + return; + } + for (PartitionEngine partition : engines.values()) { + if (!org.mockito.Mockito.mockingDetails(partition).isMock()) { + partition.shutdown(); + } + } + engines.clear(); + for (Map.Entry field : originalFields.entrySet()) { + setField(engine, field.getKey(), field.getValue()); + } + } + + @Test + public void testRestoresStaleDerivedGroupWithoutReplacingDurableRaftConfiguration() + throws Exception { + writeDurableSelfConfiguration(); + + manager.loadPartition(); + assertTrue(localGroup.get().getShardsList().stream().anyMatch(s -> s.getStoreId() == SELF)); + engine.restoreLocalPartitionEngine(); + + PartitionEngine restored = engine.getPartitionEngine(GROUP); + assertNotNull("Stale metadata must not filter out the persisted group", restored); + assertNotNull(restored.waitForLeader(5000)); + awaitLeaderPublication(); + // PD's bootstrap includes another peer; JRaft must retain its durable one-voter config. + assertEquals(List.of(JRaftUtils.getPeerId(SELF_ADDRESS)), restored.getRaftNode().listPeers()); + assertTrue(manager.hasPartition("graph", GROUP)); + verify(wrapper, never()).delete(anyInt(), any(byte[].class)); + } + + @Test + public void testSharedGroupRefreshesOnce() throws Exception { + persisted.add(partition("other-graph")); + + manager.loadPartition(); + manager.syncPartitionsFromPD(p -> { + throw new AssertionError("A current member must not be removed"); + }); + + assertTrue(manager.hasPartition("graph", GROUP)); + assertTrue(manager.hasPartition("other-graph", GROUP)); + verify(pd, times(1)).getShardGroupDirect(GROUP); + verify(wrapper, times(1)).put(eq(GROUP), eq(MetadataKeyHelper.getShardGroupKey(GROUP)), + any(byte[].class)); + } + + @Test + public void testNullDirectGroupPreservesMetadata() { + directGroup = null; + assertRecoveryBlocked(); + } + + @Test + public void testPdFailurePreservesMetadata() { + when(pd.getShardGroupDirect(GROUP)).thenThrow(new IllegalStateException("PD unavailable")); + assertRecoveryBlocked(); + } + + @Test + public void testCorruptDirectGroupsPreserveMetadata() { + for (Metapb.ShardGroup corrupt : List.of(group(SELF).toBuilder().setId(1).build(), + group(), group(0L), group(SELF, SELF), group(SELF).toBuilder() + .setShards(0, Metapb.Shard.newBuilder().setStoreId(SELF) + .setRole(Metapb.ShardRole.None)).build())) { + directGroup = corrupt; + assertRecoveryBlocked(); + } + } + + @Test + public void testMissingOrMismatchedPdPartitionPreservesMetadata() throws Exception { + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenReturn(null); + assertRecoveryBlocked(); + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenReturn(new Partition(partition("wrong"))); + assertRecoveryBlocked(); + } + + @Test + public void testPdPartitionFailurePreservesMetadata() throws Exception { + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenThrow(new IllegalStateException("PD unavailable")); + assertRecoveryBlocked(); + } + + @Test + public void testRunningEngineIsNotOverwritten() { + PartitionEngine running = mock(PartitionEngine.class); + ShardGroup observed = ShardGroup.from(group(SELF)); + when(running.getShardGroup()).thenReturn(observed); + engines.put(GROUP, running); + Metapb.ShardGroup original = localGroup.get(); + + manager.loadPartition(); + + assertSame(original, localGroup.get()); + assertSame(observed, running.getShardGroup()); + verify(wrapper, never()).put(eq(GROUP), eq(MetadataKeyHelper.getShardGroupKey(GROUP)), + any(byte[].class)); + } + + @Test + public void testPdExclusionBlocksRecoveryWithoutRetiringData() { + directGroup = group(1L); + assertRecoveryBlocked(); + assertFalse(manager.hasPartition("graph", GROUP)); + } + + @Test + public void testNoPersistentPartitionDoesNotRefreshGroup() { + persisted.clear(); + manager.loadPartition(); + verify(pd, never()).getShardGroupDirect(GROUP); + verify(wrapper, never()).put(anyInt(), any(byte[].class), any(byte[].class)); + } + + @Test + public void testLearnerWithoutLeaderCanRecoverStartupMetadata() { + directGroup = group(SELF).toBuilder().setShards(0, Metapb.Shard.newBuilder() + .setStoreId(SELF).setRole(Metapb.ShardRole.Learner)).build(); + manager.loadPartition(); + assertEquals(directGroup, localGroup.get()); + + List peers = manager.getPartitionPeers(ShardGroup.from(directGroup)); + assertEquals(List.of(SELF_ADDRESS + "::100/learner"), peers); + com.alipay.sofa.jraft.conf.Configuration initial = + new com.alipay.sofa.jraft.conf.Configuration(); + assertTrue(initial.parse(String.join(",", peers))); + assertTrue(initial.listPeers().isEmpty()); + assertEquals(SELF_ADDRESS, initial.listLearners().get(0).getEndpoint().toString()); + } + + @Test + public void testDefaultProviderPreservesRecoveryLookupErrors() throws Exception { + PDClient client = mock(PDClient.class); + DefaultPdProvider provider = mock(DefaultPdProvider.class, CALLS_REAL_METHODS); + Field clientField = DefaultPdProvider.class.getDeclaredField("pdClient"); + clientField.setAccessible(true); + clientField.set(provider, client); + PDException missing = new PDException(Pdpb.ErrorType.NOT_FOUND_VALUE, "Graph was dropped"); + when(client.getPartitionByIdDirect("graph", GROUP)).thenThrow(missing); + assertSame(missing, assertThrows(PDException.class, + () -> provider.getPartitionByIDForRecovery("graph", GROUP))); + PDException failed = new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, "RPC failed"); + doThrow(failed).when(client).getPartitionByIdDirect("graph", GROUP); + assertSame(failed, assertThrows(PDException.class, + () -> provider.getPartitionByIDForRecovery("graph", GROUP))); + doReturn(new KVPair<>(partition("graph"), Metapb.Shard.getDefaultInstance())) + .when(client).getPartitionByIdDirect("graph", GROUP); + assertEquals("graph", provider.getPartitionByIDForRecovery("graph", GROUP).getGraphName()); + } + + @Test + public void testDroppedGraphDoesNotBlockOtherGraphRecovery() throws Exception { + persisted.add(partition("other-graph")); + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenThrow( + new PDException(Pdpb.ErrorType.NOT_FOUND_VALUE, "Graph was dropped")); + when(pd.getPartitionsByStore(SELF)).thenReturn(List.of(new Partition(partition("other-graph")))); + writeDurableSelfConfiguration(); + engine.stateChanged(metadata.getStore(), Metapb.StoreState.Offline, Metapb.StoreState.Up); + assertNotNull(engine.getPartitionEngine(GROUP)); + assertNotNull(engine.getPartitionEngine(GROUP).waitForLeader(5000)); + awaitLeaderPublication(); + verify(heartbeat, never()).requestExit(anyInt()); + assertFalse(manager.hasPartition("graph", GROUP)); + assertTrue(manager.hasPartition("other-graph", GROUP)); + assertEquals(List.of(JRaftUtils.getPeerId(SELF_ADDRESS)), + engine.getPartitionEngine(GROUP).getRaftNode().listPeers()); + assertEquals(SELF, localGroup.get().getShards(0).getStoreId()); + assertEquals(1, localGroup.get().getShardsCount()); + assertEquals(2, persisted.size()); + verify(wrapper, never()).delete(anyInt(), any(byte[].class)); + verify(metadata, never()).removePartitionStore(anyInt()); + } + + @Test + public void testAllDroppedGraphsKeepPathsWithoutStartingEngineOrExitingStore() throws Exception { + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenThrow( + new PDException(Pdpb.ErrorType.NOT_FOUND_VALUE, "Graph was dropped")); + when(pd.getPartitionsByStore(SELF)).thenReturn(List.of()); + Metapb.ShardGroup original = localGroup.get(); + engine.stateChanged(metadata.getStore(), Metapb.StoreState.Offline, Metapb.StoreState.Up); + assertFalse(manager.hasPartition("graph", GROUP)); + assertTrue(engines.isEmpty()); + assertSame(original, localGroup.get()); + assertEquals(1, persisted.size()); + verify(heartbeat, never()).requestExit(anyInt()); + verify(wrapper, never()).put(anyInt(), any(byte[].class), any(byte[].class)); + verify(wrapper, never()).delete(anyInt(), any(byte[].class)); + verify(metadata, never()).removePartitionStore(anyInt()); + assertTrue(Files.isDirectory(data.resolve("raft/00000/log"))); + } + + @Test + public void testFailedPdLookupDoesNotBecomeDroppedGraph() throws Exception { + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenThrow( + new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, "RPC failed")); + assertRecoveryBlocked(); + } + + @Test + public void testEmptyRaftLogPreservesLearnerRoleWhenStartingNode() throws Exception { + directGroup = group(1L, SELF).toBuilder().setShards(1, Metapb.Shard.newBuilder() + .setStoreId(SELF).setRole(Metapb.ShardRole.Learner)).build(); + try (var logs = Files.list(data.resolve("raft/00000/log"))) { + assertEquals(0L, logs.count()); + } + manager.loadPartition(); + engine.restoreLocalPartitionEngine(); + PartitionEngine restored = engine.getPartitionEngine(GROUP); + assertNotNull("A persisted learner must use the normal recovery startup path", restored); + Configuration observed = ((NodeImpl) restored.getRaftNode()).getCurrentConf(); + assertFalse(observed.getPeers().contains(JRaftUtils.getPeerId(SELF_ADDRESS))); + assertEquals(List.of(JRaftUtils.getPeerId(OTHER_ADDRESS)), observed.listPeers()); + assertEquals(List.of(JRaftUtils.getPeerId(SELF_ADDRESS)), observed.listLearners()); + assertFalse(restored.getRaftNode().isLeader()); + } + + private void awaitLeaderPublication() throws InterruptedException { + // waitForLeader can return before the asynchronously dispatched onLeaderStart callback. + // Let its configuration publication pass the last Node access before tearing down the node. + assertTrue("Leader callback must publish its configuration before shutdown", + leaderPublication.await(5, TimeUnit.SECONDS)); + } + + private void writeDurableSelfConfiguration() throws Exception { + LogStorage storage = new RocksDBLogStorage(data.resolve("raft/00000/log").toString(), new RaftOptions()); + LogStorageOptions logOptions = new LogStorageOptions(); + logOptions.setGroupId("hg_0"); + logOptions.setConfigurationManager(new ConfigurationManager()); + logOptions.setLogEntryCodecFactory(LogEntryV2CodecFactory.getInstance()); + assertTrue(storage.init(logOptions)); + try { + LogEntry conf = new LogEntry(EntryType.ENTRY_TYPE_CONFIGURATION); + conf.setId(new LogId(1, 1)); + conf.setPeers(List.of(JRaftUtils.getPeerId(SELF_ADDRESS))); + assertTrue(storage.appendEntry(conf)); + } finally { + storage.shutdown(); + } + } + + private void assertRecoveryBlocked() { + Metapb.ShardGroup original = localGroup.get(); + assertThrows(PartitionManager.InvalidShardException.class, manager::loadPartition); + assertSame(original, localGroup.get()); + assertEquals(1, persisted.size()); + verify(wrapper, never()).put(anyInt(), any(byte[].class), any(byte[].class)); + verify(wrapper, never()).delete(anyInt(), any(byte[].class)); + verify(metadata, never()).removePartitionStore(anyInt()); + } + + private void replaceEngineField(String name, Object value) throws Exception { + Field field = HgStoreEngine.class.getDeclaredField(name); + field.setAccessible(true); + originalFields.put(name, field.get(engine)); + field.set(engine, value); + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + + private static Metapb.Partition partition(String graph) { + return Metapb.Partition.newBuilder().setId(GROUP).setGraphName(graph) + .setEndKey(65535).setState(Metapb.PartitionState.PState_Normal).build(); + } + + private static Metapb.ShardGroup group(long... stores) { + Metapb.ShardGroup.Builder group = Metapb.ShardGroup.newBuilder().setId(GROUP); + for (long id : stores) { + group.addShards(Metapb.Shard.newBuilder().setStoreId(id) + .setRole(Metapb.ShardRole.Follower)); + } + return group.build(); + } + + private static Store store(long id, String address) { + Store store = new Store(); + store.setId(id); + store.setRaftAddress(address); + return store; + } +} diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/RecoveryPdLookupTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/RecoveryPdLookupTest.java new file mode 100644 index 0000000000..3329511236 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/RecoveryPdLookupTest.java @@ -0,0 +1,158 @@ +/* + * 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.assertThrows; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; + +import org.apache.hugegraph.pd.client.ClientCache; +import org.apache.hugegraph.pd.client.PDClient; +import org.apache.hugegraph.pd.client.PDConfig; +import org.apache.hugegraph.pd.client.StubProxy; +import org.apache.hugegraph.pd.client.interceptor.Authentication; +import org.apache.hugegraph.pd.common.KVPair; +import org.apache.hugegraph.pd.common.PDException; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.apache.hugegraph.pd.grpc.PDGrpc; +import org.apache.hugegraph.pd.grpc.Pdpb; +import org.junit.Test; + +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; + +public class RecoveryPdLookupTest { + + @Test + public void testDirectMissingPartitionDoesNotUseCachedPartition() throws Exception { + Lookup lookup = new Lookup(); + when(lookup.cache.getPartitionById("graph", 0)).thenReturn(new KVPair<>( + Metapb.Partition.newBuilder().setGraphName("graph").build(), + Metapb.Shard.getDefaultInstance())); + lookup.response = Pdpb.GetPartitionResponse.newBuilder().setHeader( + Pdpb.ResponseHeader.newBuilder().setError( + Pdpb.Error.newBuilder().setType(Pdpb.ErrorType.NOT_FOUND))).build(); + PDException missing = assertThrows(PDException.class, + () -> lookup.client.getPartitionByIdDirect("graph", 0)); + assertEquals(Pdpb.ErrorType.NOT_FOUND_VALUE, missing.getErrorCode()); + assertEquals(1, lookup.requests.size()); + assertEquals("graph", lookup.requests.get(0).getGraphName()); + assertEquals(0, lookup.requests.get(0).getPartitionId()); + verifyNoInteractions(lookup.cache); + } + + @Test + public void testDirectPartitionDoesNotQueryOrPopulateShardGroupCache() throws Exception { + Lookup lookup = new Lookup(); + KVPair pair = lookup.client.getPartitionByIdDirect("graph", 0); + assertEquals(lookup.response.getPartition(), pair.getKey()); + assertEquals(1, lookup.requests.size()); + verifyNoInteractions(lookup.cache); + } + + @Test + public void testFailedRpcRemainsFailureAndCanBeRetried() throws Exception { + Lookup lookup = new Lookup(); + lookup.status = Status.UNAVAILABLE.withDescription("PD connection failed"); + PDException failure = assertThrows(PDException.class, + () -> lookup.client.getPartitionByIdDirect("graph", 0)); + assertEquals(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, failure.getErrorCode()); + lookup.status = Status.OK; + assertEquals("graph", lookup.client.getPartitionByIdDirect("graph", 0).getKey().getGraphName()); + assertEquals(2, lookup.requests.size()); + verifyNoInteractions(lookup.cache); + } + + private static final class Lookup extends Channel { + + private final PDClient client = mock(PDClient.class, CALLS_REAL_METHODS); + private final ClientCache cache = mock(ClientCache.class); + private final List requests = new ArrayList<>(); + private Pdpb.GetPartitionResponse response = Pdpb.GetPartitionResponse.newBuilder() + .setPartition(Metapb.Partition.newBuilder().setGraphName("graph")).build(); + private Status status = Status.OK; + + private Lookup() throws Exception { + StubProxy proxy = new StubProxy(new String[]{"unused"}); + proxy.set(PDGrpc.newBlockingStub(this)); + setField("proxy", proxy); + setField("config", PDConfig.of("unused").setEnableCache(true)); + setField("header", Pdpb.RequestHeader.getDefaultInstance()); + setField("cache", cache); + setField("auth", new Authentication("store", "test")); + } + + private void setField(String name, Object value) throws Exception { + Field field = PDClient.class.getDeclaredField(name); + field.setAccessible(true); + field.set(client, value); + } + + @Override + public String authority() { + return "recovery-test"; + } + + @Override + public ClientCall newCall(MethodDescriptor method, + CallOptions options) { + assertEquals(PDGrpc.getGetPartitionByIDMethod(), method); + return new ClientCall<>() { + private Listener listener; + + @Override + public void start(Listener listener, Metadata headers) { + this.listener = listener; + } + + @Override + public void request(int numMessages) { + } + + @Override + public void cancel(String message, Throwable cause) { + } + + @Override + @SuppressWarnings("unchecked") + public void halfClose() { + if (status.isOk()) { + listener.onMessage((RespT) response); + } + listener.onClose(status, new Metadata()); + } + + @Override + public void sendMessage(ReqT request) { + requests.add((Pdpb.GetPartitionByIDRequest) request); + } + }; + } + } +}