From da2de4d17f4be20e121f1f52ba7ff69700d31c70 Mon Sep 17 00:00:00 2001 From: orion Date: Tue, 27 Oct 2026 09:00:25 +0000 Subject: [PATCH 1/3] fix(store): refresh recovery shard metadata - Validate persisted groups against direct PD membership before startup recovery. - Refresh derived shard metadata only before its engine starts, retaining durable Raft configuration. - Cover persisted configuration recovery and refusal guards, and document restart checks. --- hugegraph-store/docs/operations-guide.md | 15 + .../store/meta/PartitionManager.java | 104 ++++-- hugegraph-store/hg-store-test/pom.xml | 1 + .../store/HeartbeatShutdownTest.java | 2 +- .../PartitionRecoveryMembershipTest.java | 352 ++++++++++++++++++ 5 files changed, 434 insertions(+), 40 deletions(-) create mode 100644 hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionRecoveryMembershipTest.java diff --git a/hugegraph-store/docs/operations-guide.md b/hugegraph-store/docs/operations-guide.md index 78e1c718b5..01dbca5b0a 100644 --- a/hugegraph-store/docs/operations-guide.md +++ b/hugegraph-store/docs/operations-guide.md @@ -654,6 +654,21 @@ curl http://192.168.1.10:8620/v1/partitionsAndStatus 5. **Verify**: Wait for even distribution (may take hours) +### 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. + +Missing, invalid or unavailable PD metadata blocks recovery while retaining local data; +this includes members with an unspecified `None` role until PD reports usable member roles. +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..30b59a3209 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 @@ -107,6 +107,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() { @@ -269,53 +273,50 @@ 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(); + } 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; + Partition pdPartition; + try { + pdPartition = pdProvider.getPartitionByID(graph, metaPart.getId()); + } catch (RuntimeException e) { + throw new InvalidShardException("Cannot read persisted partition " + graph + + "-" + partId + " from PD", e); } - - if (isLegeal) { - if (!partitions.containsKey(graph)) { - partitions.put(graph, new ConcurrentHashMap<>()); - } - - 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(); + if (metaPart.getId() != partId || graph.isEmpty() || pdPartition == null || + pdPartition.getId() != partId || !graph.equals(pdPartition.getGraphName())) { + throw new InvalidShardException("Cannot verify persisted partition " + graph + + "-" + partId + " with PD", null); } } - 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 : metaParts) { + 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); } @@ -327,6 +328,31 @@ private void loadPartitions() { } } + 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); diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index 28ca7102bf..e67ef886f4 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 + **/PartitionRecoveryMembershipTest.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..964fbe4a1a 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,7 +192,7 @@ 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)); 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..979b176f5e --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/PartitionRecoveryMembershipTest.java @@ -0,0 +1,352 @@ +/* + * 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.doAnswer; +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.atomic.AtomicReference; + +import org.apache.hugegraph.pd.grpc.Metapb; +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.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.ConfigurationManager; +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; + + @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); + when(pd.getPartitionByID(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)); + replaceEngineField("heartbeatService", mock(HeartbeatService.class)); + 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 { + 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(); + } + + 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)); + // 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() { + when(pd.getPartitionByID("graph", GROUP)).thenReturn(null); + assertRecoveryBlocked(); + when(pd.getPartitionByID("graph", GROUP)).thenReturn(new Partition(partition("wrong"))); + assertRecoveryBlocked(); + } + + @Test + public void testPdPartitionFailurePreservesMetadata() { + when(pd.getPartitionByID("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()); + } + + 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; + } +} From 477220821a117fdaeb1fb89e42104ed182513b2a Mon Sep 17 00:00:00 2001 From: imbajin Date: Wed, 28 Oct 2026 04:21:43 +0800 Subject: [PATCH 2/3] fix(store): preserve learner role in recovery bootstrap --- .../org/apache/hugegraph/store/meta/PartitionManager.java | 6 +++++- .../hugegraph/store/PartitionRecoveryMembershipTest.java | 8 ++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) 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 30b59a3209..098e1cf436 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 @@ -823,7 +823,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-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 index 979b176f5e..ea729a230b 100644 --- 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 @@ -304,6 +304,14 @@ public void testLearnerWithoutLeaderCanRecoverStartupMetadata() { .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()); } private void assertRecoveryBlocked() { From f005ef65d8d417ab7f62ce3728694678520583d6 Mon Sep 17 00:00:00 2001 From: orion Date: Wed, 28 Oct 2026 05:14:28 +0000 Subject: [PATCH 3/3] fix(store): skip dropped graphs during recovery Preserve direct PD lookup errors and distinguish graph absence from failed RPCs. Retain skipped graph data and paths while recovering remaining valid graphs. Verify native learner startup and await leader callback publication in recovery tests. --- .../apache/hugegraph/pd/client/PDClient.java | 25 +-- hugegraph-store/docs/operations-guide.md | 10 +- .../store/meta/PartitionManager.java | 34 +++- .../hugegraph/store/pd/DefaultPdProvider.java | 6 + .../apache/hugegraph/store/pd/PdProvider.java | 8 + hugegraph-store/hg-store-test/pom.xml | 1 + .../store/HeartbeatShutdownTest.java | 2 +- .../PartitionRecoveryMembershipTest.java | 165 +++++++++++++++--- .../hugegraph/store/RecoveryPdLookupTest.java | 158 +++++++++++++++++ 9 files changed, 369 insertions(+), 40 deletions(-) create mode 100644 hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/RecoveryPdLookupTest.java 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 01dbca5b0a..5e4452b079 100644 --- a/hugegraph-store/docs/operations-guide.md +++ b/hugegraph-store/docs/operations-guide.md @@ -662,8 +662,14 @@ engine is running for the group. This lets JRaft restore its durable configurati when the business database contains older membership metadata. Running engines and durable Raft configuration are not replaced by the PD bootstrap metadata. -Missing, invalid or unavailable PD metadata blocks recovery while retaining local data; -this includes members with an unspecified `None` role until PD reports usable member roles. +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 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 098e1cf436..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; @@ -263,6 +264,7 @@ private void loadPartitions() { } Set normalPartitions = new HashSet<>(); + Set retainedPartitions = new HashSet<>(); // Once according to the partition read for (int partId : partIds) { @@ -287,20 +289,41 @@ private void loadPartitions() { partId, storeId, shards2Peers(shards)); throw new InvalidShardException(); } + List recoverableParts = new ArrayList<>(); for (var metaPart : metaParts) { var graph = metaPart.getGraphName(); + if (metaPart.getId() != partId || graph.isEmpty()) { + throw new InvalidShardException("Invalid persisted partition " + graph + + "-" + partId, null); + } Partition pdPartition; try { - pdPartition = pdProvider.getPartitionByID(graph, metaPart.getId()); + 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; + } + 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 (metaPart.getId() != partId || graph.isEmpty() || pdPartition == null || - pdPartition.getId() != partId || !graph.equals(pdPartition.getGraphName())) { + 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; } // Only refresh startup metadata, once per group even when several graphs share it. @@ -308,7 +331,7 @@ private void loadPartitions() { if (HgStoreEngine.getInstance().getPartitionEngine(partId) == null) { updateShardGroup(ShardGroup.from(pdGroup)); } - for (var metaPart : metaParts) { + for (var metaPart : recoverableParts) { var graph = metaPart.getGraphName(); partitions.computeIfAbsent(graph, g -> new ConcurrentHashMap<>()); Partition partition = new Partition(metaPart); @@ -322,7 +345,8 @@ private void loadPartitions() { // 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()); } } 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 e67ef886f4..54e16bee48 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -247,6 +247,7 @@ **/PartitionStateMachineTest.java **/PartitionEngineErrorTest.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 964fbe4a1a..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 @@ -195,7 +195,7 @@ private static void runInvalidShardRecovery(Path marker) throws Exception { 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 index ea729a230b..4c113ca647 100644 --- 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 @@ -28,7 +28,10 @@ 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; @@ -43,9 +46,15 @@ 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; @@ -55,6 +64,7 @@ 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; @@ -64,7 +74,9 @@ 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; @@ -97,6 +109,8 @@ public class PartitionRecoveryMembershipTest { private Metapb.ShardGroup directGroup; private Path data; private Map engines; + private HeartbeatService heartbeat; + private CountDownLatch leaderPublication; @Before @SuppressWarnings("unchecked") @@ -142,7 +156,17 @@ public void setUp() throws Exception { }).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); - when(pd.getPartitionByID(anyString(), eq(GROUP))) + 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<>(); @@ -161,7 +185,8 @@ public void setUp() throws Exception { replaceEngineField("options", options); replaceEngineField("pdProvider", pd); replaceEngineField("businessHandler", mock(BusinessHandler.class)); - replaceEngineField("heartbeatService", mock(HeartbeatService.class)); + heartbeat = mock(HeartbeatService.class); + replaceEngineField("heartbeatService", heartbeat); RpcServer rpc = mock(RpcServer.class); when(rpc.boundPort()).thenReturn(18539); replaceEngineField("rpcServer", rpc); @@ -186,20 +211,7 @@ public void tearDown() throws Exception { @Test public void testRestoresStaleDerivedGroupWithoutReplacingDurableRaftConfiguration() 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(); - } + writeDurableSelfConfiguration(); manager.loadPartition(); assertTrue(localGroup.get().getShardsList().stream().anyMatch(s -> s.getStoreId() == SELF)); @@ -208,6 +220,7 @@ public void testRestoresStaleDerivedGroupWithoutReplacingDurableRaftConfiguratio 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)); @@ -254,16 +267,16 @@ public void testCorruptDirectGroupsPreserveMetadata() { } @Test - public void testMissingOrMismatchedPdPartitionPreservesMetadata() { - when(pd.getPartitionByID("graph", GROUP)).thenReturn(null); + public void testMissingOrMismatchedPdPartitionPreservesMetadata() throws Exception { + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenReturn(null); assertRecoveryBlocked(); - when(pd.getPartitionByID("graph", GROUP)).thenReturn(new Partition(partition("wrong"))); + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenReturn(new Partition(partition("wrong"))); assertRecoveryBlocked(); } @Test - public void testPdPartitionFailurePreservesMetadata() { - when(pd.getPartitionByID("graph", GROUP)).thenThrow(new IllegalStateException("PD unavailable")); + public void testPdPartitionFailurePreservesMetadata() throws Exception { + when(pd.getPartitionByIDForRecovery("graph", GROUP)).thenThrow(new IllegalStateException("PD unavailable")); assertRecoveryBlocked(); } @@ -314,6 +327,116 @@ public void testLearnerWithoutLeaderCanRecoverStartupMetadata() { 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); 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); + } + }; + } + } +}