diff --git a/hugegraph-pd/README.md b/hugegraph-pd/README.md index a5dc5d6f13..5d62d76659 100644 --- a/hugegraph-pd/README.md +++ b/hugegraph-pd/README.md @@ -291,6 +291,7 @@ docker/docker-compose-3pd-3store-3server.yml - PD uses Raft consensus for leader election and data replication - Cluster can tolerate up to `(N-1)/2` node failures (e.g., 1 failure in 3-node cluster) - Leader handles all write operations; followers handle read operations +- Never delete RocksDB `LOCK` files under `pd.data-path` to get a restarted PD going; see [PD Core Settings](docs/configuration.md#pd-core-settings) ### Network Requirements diff --git a/hugegraph-pd/docs/api-reference.md b/hugegraph-pd/docs/api-reference.md index 35d70fadd4..122ae6c8b4 100644 --- a/hugegraph-pd/docs/api-reference.md +++ b/hugegraph-pd/docs/api-reference.md @@ -813,8 +813,8 @@ Two unauthenticated endpoints are meant for probes and startup gates: | `GET /v1/ready` | Readiness: the raft node is active and sees a leader, so this PD is inside a quorum. | `200` when ready, `503` otherwise | The health probe reads Raft state only; it does not verify metadata-store initialization. -Metadata open failures propagate through startup; bounded LOCK retry remains tracked in -[#3226](https://github.com/apache/hugegraph/issues/3226). +If the metadata store cannot be opened, PD startup fails; while its RocksDB `LOCK` is still +held, PD first retries for up to 30 seconds (see [PD Core Settings](configuration.md#pd-core-settings)). ```bash curl -i http://localhost:8620/v1/ready diff --git a/hugegraph-pd/docs/configuration.md b/hugegraph-pd/docs/configuration.md index c2d6eecc9f..18cd0dd884 100644 --- a/hugegraph-pd/docs/configuration.md +++ b/hugegraph-pd/docs/configuration.md @@ -194,6 +194,8 @@ pd: **Production Recommendations**: - `pd.data-path`: Use dedicated SSD with at least 50GB free space + - PD opens its metadata RocksDB at `/rocksdb`. If its `LOCK` is still held, usually by a previous PD process that is still stopping, PD retries for up to 30 seconds and then fails startup. Run PD in the foreground (`-d false`) under a supervisor so that a failed start exits and is restarted. + - Never delete the `LOCK` files in `/rocksdb/` or `/pd_raft/log/` to get past this: while the previous process is alive, two processes then write the same database. Stop or kill the old process instead. - `pd.patrol-interval`: - Development: `300` (5 minutes) for fast testing - Production: `1800` (30 minutes) to reduce overhead diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/meta/MetadataFactory.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/meta/MetadataFactory.java index 954a9f9e2e..e7c64d0222 100644 --- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/meta/MetadataFactory.java +++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/meta/MetadataFactory.java @@ -56,6 +56,26 @@ public static synchronized HgKVStore getStore(PDConfig pdConfig) { return store; } + /** + * Open the store like {@link #getStore}, and return it only when this call opened it. + */ + public static synchronized HgKVStore openStore(PDConfig pdConfig) { + HgKVStore previous = store; + HgKVStore current = getStore(pdConfig); + return previous == null ? current : null; + } + + /** + * Close a store that a failed startup opened, so its LOCK does not outlive the attempt. + * A store another caller has replaced it with stays open. + */ + public static synchronized void discardStore(HgKVStore opened) { + if (opened != null && store == opened) { + store = null; + opened.close(); + } + } + public static synchronized void closeStore() { if (store != null) { store.close(); diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/service/RaftStartup.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/service/RaftStartup.java new file mode 100644 index 0000000000..53b27c8a14 --- /dev/null +++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/service/RaftStartup.java @@ -0,0 +1,62 @@ +/* + * 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.pd.service; + +import org.apache.hugegraph.pd.config.PDConfig; +import org.apache.hugegraph.pd.meta.MetadataFactory; +import org.apache.hugegraph.pd.raft.RaftEngine; +import org.apache.hugegraph.pd.store.HgKVStore; + +/** + * Starts the shared {@link RaftEngine} from any service's startup callback. + */ +public final class RaftStartup { + + private RaftStartup() { + } + + /** + * Open the metadata store before Raft. Its open waits for a LOCK still held by a stopping + * PD, and that PD closes Raft before metadata, so the raft log LOCK is free once the + * metadata store is open. Spring runs the service callbacks in no fixed order, so each of + * them starts Raft through here. If Raft then fails to start, a metadata store this call + * opened is closed again, since a callback other than PDService has no owner to close it. + */ + public static void init(PDConfig pdConfig) { + HgKVStore opened = MetadataFactory.openStore(pdConfig); + try { + RaftEngine.getInstance().init(pdConfig.getRaft()); + } catch (RuntimeException | Error failure) { + // Only a store this call opened is ours to release; PDService and the + // shutdown hook own the rest + if (opened != null) { + try { + // Join Raft before closing metadata, as on shutdown; a failed join + // keeps the database open + RaftEngine.getInstance().shutDown(); + MetadataFactory.discardStore(opened); + } catch (RuntimeException | Error cleanup) { + if (cleanup != failure) { + failure.addSuppressed(cleanup); + } + } + } + throw failure; + } + } +} diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/HgKVStoreImpl.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/HgKVStoreImpl.java index f33648e0eb..105f54736d 100644 --- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/HgKVStoreImpl.java +++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/HgKVStoreImpl.java @@ -43,6 +43,7 @@ import org.rocksdb.RocksDBException; import org.rocksdb.RocksIterator; import org.rocksdb.Slice; +import org.rocksdb.Status; import com.alipay.sofa.jraft.util.Utils; import com.google.common.cache.CacheBuilder; @@ -55,10 +56,22 @@ public class HgKVStoreImpl implements HgKVStore { private static final ConcurrentHashMap> CACHE = new ConcurrentHashMap(); + /* + * A restarted PD can start before the previous JVM has released the RocksDB LOCK. + * Wait about as long as stop-hugegraph-pd.sh waits for that JVM, then fail startup. + * Every attempt rotates the RocksDB info LOG, so back off between attempts. Only the + * metadata store waits: RaftStartup opens it before the raft log store, and a stopping + * PD releases the raft log LOCK before this one. + */ + private static final long LOCK_WAIT_TIMEOUT_MS = 30_000L; + private static final long LOCK_RETRY_MIN_INTERVAL_MS = 500L; + private static final long LOCK_RETRY_MAX_INTERVAL_MS = 4_000L; private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); private RocksDB db; private String dbPath; private Options dbOptions; + // Tests shorten this + private long lockWaitTimeoutMs = LOCK_WAIT_TIMEOUT_MS; @Override public void init(PDConfig config) { @@ -75,10 +88,8 @@ public void init(PDConfig config) { log.warn("Failed to create data file,{}", e); } } - openRocksDB(dbPath); + openRocksDB(this.dbPath, this.lockWaitTimeoutMs); } catch (PDException | RuntimeException | Error e) { - // TODO: retry opening a held RocksDB LOCK before failing PD startup. - // https://github.com/apache/hugegraph/issues/3226 if (this.dbOptions != null) { this.dbOptions.close(); this.dbOptions = null; @@ -348,11 +359,55 @@ private void closeRocksDB() { } private void openRocksDB(String dbPath) throws PDException { - try { - this.db = RocksDB.open(dbOptions, dbPath); - } catch (RocksDBException e) { - log.error("Failed to open RocksDB from {}", dbPath, e); - throw new PDException(Pdpb.ErrorType.ROCKSDB_LOAD_SNAPSHOT_ERROR_VALUE, e); + openRocksDB(dbPath, 0L); + } + + private void openRocksDB(String dbPath, long lockWaitTimeoutMs) throws PDException { + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(lockWaitTimeoutMs); + long interval = LOCK_RETRY_MIN_INTERVAL_MS; + boolean waited = false; + while (true) { + try { + this.db = RocksDB.open(this.dbOptions, dbPath); + if (waited) { + log.info("Opened RocksDB at {} after its LOCK was released", dbPath); + } + return; + } catch (RocksDBException e) { + if (!isLockConflict(e) || System.nanoTime() - deadline >= 0) { + log.error("Failed to open RocksDB from {}", dbPath, e); + throw new PDException(Pdpb.ErrorType.ROCKSDB_LOAD_SNAPSHOT_ERROR_VALUE, e); + } + if (!waited) { + log.warn("RocksDB LOCK at {} is held, probably by a previous PD that is " + + "still stopping; retrying for up to {} ms: {}", + dbPath, lockWaitTimeoutMs, e.getMessage()); + waited = true; + } + try { + long remainingMs = TimeUnit.NANOSECONDS.toMillis(deadline - System.nanoTime()); + Thread.sleep(Math.max(1L, Math.min(interval, remainingMs))); + interval = Math.min(interval * 2, LOCK_RETRY_MAX_INTERVAL_MS); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + e.addSuppressed(ie); + throw new PDException(Pdpb.ErrorType.ROCKSDB_LOAD_SNAPSHOT_ERROR_VALUE, e); + } + } } } + + /** + * The LOCK is held by another process ("While lock file") or by another handle in this + * JVM ("lock hold by current process"). "While lock file" also covers a file system that + * cannot lock at all; that fails once the wait runs out. Other open failures, including a + * LOCK file that cannot be created, are not retried. + */ + private static boolean isLockConflict(RocksDBException e) { + Status status = e.getStatus(); + String message = e.getMessage(); + return status != null && status.getCode() == Status.Code.IOError && message != null && + (message.contains("While lock file") || + message.contains("lock hold by current process")); + } } diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/rest/StoreAPI.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/rest/StoreAPI.java index e4fea9d16e..a7ad2a33f8 100644 --- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/rest/StoreAPI.java +++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/rest/StoreAPI.java @@ -395,8 +395,7 @@ class StoreStatistics { * @return Returns a string indicating the service's health status. Typically, an empty * string indicates the service is healthy. */ - // TODO: bounded metadata-store LOCK retry is tracked separately in - // https://github.com/apache/hugegraph/issues/3226; this probe reads only Raft state. + // This probe reads only Raft state; metadata-store open failures fail PD startup instead. @GetMapping(value = "/health", produces = MediaType.TEXT_PLAIN_VALUE) public ResponseEntity checkHealthy() { String state = RaftEngine.getInstance().getRaftStatus().getState(); diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/DiscoveryService.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/DiscoveryService.java index 00a652577f..e74b529c91 100644 --- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/DiscoveryService.java +++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/DiscoveryService.java @@ -59,7 +59,7 @@ public class DiscoveryService extends DiscoveryServiceGrpc.DiscoveryServiceImplB @PostConstruct public void init() throws PDException { log.info("PDService init………… {}", pdConfig); - RaftEngine.getInstance().init(pdConfig.getRaft()); + RaftStartup.init(pdConfig); RaftEngine.getInstance().addStateListener(this); register = new RegistryService(pdConfig); licenseVerifierService = new LicenseVerifierService(pdConfig); diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java index 94b41a0f0a..8d9438d391 100644 --- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java +++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java @@ -75,7 +75,7 @@ public class KvServiceGrpcImpl extends KvServiceGrpc.KvServiceImplBase implement @PostConstruct public void init() { - RaftEngine.getInstance().init(pdConfig.getRaft()); + RaftStartup.init(pdConfig); RaftEngine.getInstance().addStateListener(this); kvService = new KvService(pdConfig); subjects = new KvWatchSubject(pdConfig); diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/PDService.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/PDService.java index f40b2f5801..500bd41959 100644 --- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/PDService.java +++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/PDService.java @@ -199,7 +199,7 @@ private void initService() throws PDException { RaftEngine.getInstance().addStateListener(this); RaftEngine.getInstance().addStateListener(configService); - RaftEngine.getInstance().init(pdConfig.getRaft()); + RaftStartup.init(pdConfig); //pdConfig = configService.loadConfig(); onLeaderChanged storeNodeService = new StoreNodeService(pdConfig); partitionService = new PartitionService(pdConfig, storeNodeService); diff --git a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/PDCoreSuiteTest.java b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/PDCoreSuiteTest.java index c0c15975e8..ca0b94adae 100644 --- a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/PDCoreSuiteTest.java +++ b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/PDCoreSuiteTest.java @@ -44,6 +44,7 @@ IdMetaStoreReadIndexTest.class, HgKVStoreImplTest.class, PDNativeOwnerTest.class, + RaftStartupTest.class, PDConfigTest.class, ConfigServiceTest.class, IdServiceTest.class, diff --git a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/RaftStartupTest.java b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/RaftStartupTest.java new file mode 100644 index 0000000000..8e6a441cae --- /dev/null +++ b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/RaftStartupTest.java @@ -0,0 +1,135 @@ +/* + * 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.pd.core; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import java.io.File; +import java.lang.reflect.Field; + +import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.pd.config.PDConfig; +import org.apache.hugegraph.pd.meta.MetadataFactory; +import org.apache.hugegraph.pd.raft.RaftEngine; +import org.apache.hugegraph.pd.service.RaftStartup; +import org.apache.hugegraph.pd.store.HgKVStore; +import org.junit.Assert; +import org.junit.Test; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; + +public class RaftStartupTest { + + @Test + public void testRaftDoesNotStartBeforeMetadataOpens() throws Exception { + String dataPath = "tmp/test-raft-startup"; + PDConfig config = newConfig(dataPath, "127.0.0.1:0"); + // A regular file where the metadata RocksDB directory belongs makes its open fail + FileUtils.forceMkdir(new File(dataPath)); + FileUtils.touch(new File(dataPath, "rocksdb")); + withFactoryStore(null, () -> { + IllegalStateException e = Assert.assertThrows(IllegalStateException.class, + () -> RaftStartup.init(config)); + Assert.assertTrue(e.getMessage(), + e.getMessage().startsWith("Failed to open PD metadata")); + Assert.assertFalse(new File(dataPath, "raft").exists()); + Assert.assertNull(RaftEngine.getInstance().getRaftNode()); + }); + } + + @Test + public void testRaftFailureReleasesMetadataItOpened() throws Exception { + String dataPath = "tmp/test-raft-startup-owned"; + // An address jraft cannot parse makes Raft fail after metadata has opened + PDConfig config = newConfig(dataPath, "not-a-peer"); + withFactoryStore(null, () -> { + Assert.assertThrows(IllegalArgumentException.class, () -> RaftStartup.init(config)); + Assert.assertNull(factoryStore()); + Assert.assertNull(RaftEngine.getInstance().getRaftNode()); + // The metadata LOCK is free again + try (Options options = new Options(); + RocksDB ignored = RocksDB.open(options, dataPath + "/rocksdb")) { + Assert.assertNotNull(ignored); + } + }); + } + + @Test + public void testRaftFailureKeepsMetadataOpenedByAnotherOwner() throws Exception { + PDConfig config = newConfig("tmp/test-raft-startup-shared", "not-a-peer"); + HgKVStore shared = mock(HgKVStore.class); + withFactoryStore(shared, () -> { + Assert.assertThrows(IllegalArgumentException.class, () -> RaftStartup.init(config)); + Assert.assertSame(shared, factoryStore()); + verify(shared, never()).close(); + }); + } + + private static PDConfig newConfig(String dataPath, String raftAddress) throws Exception { + FileUtils.deleteQuietly(new File(dataPath)); + PDConfig config = new PDConfig(); + config.setDataPath(dataPath); + PDConfig.Raft raft = config.new Raft(); + raft.setDataPath(dataPath + "/raft"); + raft.setAddress(raftAddress); + config.setRaft(raft); + return config; + } + + private static Object factoryStore() throws Exception { + return factoryField("store").get(null); + } + + private static Field factoryField(String name) throws Exception { + Field field = MetadataFactory.class.getDeclaredField(name); + field.setAccessible(true); + return field; + } + + private static void withFactoryStore(HgKVStore initial, Body body) throws Exception { + Assert.assertNull(RaftEngine.getInstance().getRaftNode()); + Field storeField = factoryField("store"); + Field shutdownField = factoryField("shutdown"); + synchronized (MetadataFactory.class) { + Object previous = storeField.get(null); + boolean shutdown = shutdownField.getBoolean(null); + try { + storeField.set(null, initial); + shutdownField.setBoolean(null, false); + body.run(); + } finally { + Object left = storeField.get(null); + if (left != null && left != initial) { + ((HgKVStore) left).close(); + } + storeField.set(null, previous); + shutdownField.setBoolean(null, shutdown); + FileUtils.deleteQuietly(new File("tmp/test-raft-startup")); + FileUtils.deleteQuietly(new File("tmp/test-raft-startup-owned")); + FileUtils.deleteQuietly(new File("tmp/test-raft-startup-shared")); + } + } + } + + private interface Body { + + void run() throws Exception; + } +} diff --git a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/HgKVStoreImplTest.java b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/HgKVStoreImplTest.java index 36ef499038..772f7a7909 100644 --- a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/HgKVStoreImplTest.java +++ b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/HgKVStoreImplTest.java @@ -17,9 +17,13 @@ package org.apache.hugegraph.pd.core.store; +import java.io.BufferedReader; import java.io.File; import java.io.IOException; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; import java.nio.file.Paths; +import java.util.concurrent.TimeUnit; import org.apache.commons.io.FileUtils; import org.apache.hugegraph.pd.common.PDException; @@ -28,9 +32,13 @@ import org.apache.hugegraph.pd.meta.MetadataKeyHelper; import org.apache.hugegraph.pd.store.HgKVStore; import org.apache.hugegraph.pd.store.HgKVStoreImpl; +import org.apache.hugegraph.testutil.Whitebox; import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; +import org.rocksdb.RocksDBException; public class HgKVStoreImplTest { @@ -135,4 +143,159 @@ public void TestSnapshot() throws PDException { kvStore.close(); } + + @Test + public void testInitWaitsForHeldLock() throws Exception { + PDConfig config = newConfig("tmp/test-lock-released"); + String dbPath = config.getDataPath() + "/rocksdb"; + FileUtils.forceMkdir(new File(dbPath)); + HgKVStore kvStore = new HgKVStoreImpl(); + try (Options options = new Options().setCreateIfMissing(true)) { + RocksDB holder = RocksDB.open(options, dbPath); + // Release the LOCK the way a previous PD JVM does while it is still stopping + Thread releaser = new Thread(() -> { + try { + Thread.sleep(1000L); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + } + holder.close(); + }, "test-lock-releaser"); + releaser.setDaemon(true); + releaser.start(); + try { + long start = System.nanoTime(); + kvStore.init(config); + long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + Assert.assertTrue("opened after " + elapsed + " ms", elapsed >= 500L); + byte[] key = "lock".getBytes(StandardCharsets.UTF_8); + kvStore.put(key, key); + Assert.assertArrayEquals(key, kvStore.get(key)); + } finally { + releaser.join(); + kvStore.close(); + FileUtils.deleteQuietly(new File(config.getDataPath())); + } + } + } + + @Test + public void testInitFailsWhenLockIsNeverReleased() throws Exception { + PDConfig config = newConfig("tmp/test-lock-held"); + String dbPath = config.getDataPath() + "/rocksdb"; + FileUtils.forceMkdir(new File(dbPath)); + HgKVStoreImpl kvStore = new HgKVStoreImpl(); + Whitebox.setInternalState(kvStore, "lockWaitTimeoutMs", 1000L); + try (Options options = new Options().setCreateIfMissing(true); + RocksDB ignored = RocksDB.open(options, dbPath)) { + long start = System.nanoTime(); + IllegalStateException e = Assert.assertThrows(IllegalStateException.class, + () -> kvStore.init(config)); + long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + Assert.assertTrue("gave up after " + elapsed + " ms", elapsed >= 1000L); + Assert.assertTrue(e.getCause() instanceof PDException); + Assert.assertTrue(e.getCause().getCause() instanceof RocksDBException); + } finally { + kvStore.close(); + FileUtils.deleteQuietly(new File(config.getDataPath())); + } + } + + @Test + public void testInitDoesNotRetryOtherOpenFailures() throws Exception { + PDConfig config = newConfig("tmp/test-lock-not-a-dir"); + // A regular file where the RocksDB directory belongs is not a lock conflict + File dbFile = new File(config.getDataPath() + "/rocksdb"); + FileUtils.forceMkdir(dbFile.getParentFile()); + FileUtils.touch(dbFile); + HgKVStore kvStore = new HgKVStoreImpl(); + long start = System.nanoTime(); + try { + IllegalStateException e = Assert.assertThrows(IllegalStateException.class, + () -> kvStore.init(config)); + String message = e.getCause().getCause().getMessage(); + Assert.assertTrue(message, message.contains("is not a directory")); + } finally { + kvStore.close(); + FileUtils.deleteQuietly(new File(config.getDataPath())); + } + long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + Assert.assertTrue("retried for " + elapsed + " ms", elapsed < 5000L); + } + + @Test(timeout = 60_000L) + public void testInitWaitsForLockHeldByAnotherProcess() throws Exception { + PDConfig config = newConfig("tmp/test-process-lock-released"); + String dbPath = config.getDataPath() + "/rocksdb"; + FileUtils.forceMkdir(new File(dbPath)); + HgKVStore kvStore = new HgKVStoreImpl(); + Process holder = startLockHolder(dbPath, 1000L); + try { + long start = System.nanoTime(); + kvStore.init(config); + long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + Assert.assertTrue("opened after " + elapsed + " ms", elapsed >= 500L); + byte[] key = "lock".getBytes(StandardCharsets.UTF_8); + kvStore.put(key, key); + Assert.assertArrayEquals(key, kvStore.get(key)); + Assert.assertEquals(0, holder.waitFor()); + } finally { + holder.destroyForcibly().waitFor(); + kvStore.close(); + FileUtils.deleteQuietly(new File(config.getDataPath())); + } + } + + @Test(timeout = 60_000L) + public void testInitFailsWhenAnotherProcessKeepsLock() throws Exception { + PDConfig config = newConfig("tmp/test-process-lock-held"); + String dbPath = config.getDataPath() + "/rocksdb"; + FileUtils.forceMkdir(new File(dbPath)); + HgKVStoreImpl kvStore = new HgKVStoreImpl(); + Whitebox.setInternalState(kvStore, "lockWaitTimeoutMs", 1000L); + Process holder = startLockHolder(dbPath, -1L); + try { + long start = System.nanoTime(); + IllegalStateException e = Assert.assertThrows(IllegalStateException.class, + () -> kvStore.init(config)); + long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + Assert.assertTrue("gave up after " + elapsed + " ms", elapsed >= 1000L); + Throwable cause = e.getCause().getCause(); + Assert.assertTrue(cause instanceof RocksDBException); + Assert.assertTrue(cause.getMessage(), cause.getMessage().contains("While lock file")); + } finally { + holder.destroyForcibly().waitFor(); + kvStore.close(); + FileUtils.deleteQuietly(new File(config.getDataPath())); + } + } + + private static Process startLockHolder(String dbPath, long holdMs) throws Exception { + String java = Paths.get(System.getProperty("java.home"), "bin", "java").toString(); + Process process = new ProcessBuilder(java, "-cp", System.getProperty("java.class.path"), + RocksDBLockHolder.class.getName(), dbPath, + String.valueOf(holdMs)) + .redirectErrorStream(true) + .start(); + BufferedReader output = new BufferedReader( + new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8)); + String line; + do { + line = output.readLine(); + } while (line != null && !RocksDBLockHolder.LOCKED.equals(line)); + if (line == null) { + Assert.fail("Lock holder exited with " + process.waitFor()); + } + return process; + } + + private static PDConfig newConfig(String path) throws IOException { + File dir = new File(path); + if (dir.exists()) { + FileUtils.deleteDirectory(dir); + } + return new PDConfig() {{ + setDataPath(path); + }}; + } } diff --git a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/RocksDBLockHolder.java b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/RocksDBLockHolder.java new file mode 100644 index 0000000000..9cdec9cedb --- /dev/null +++ b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/RocksDBLockHolder.java @@ -0,0 +1,46 @@ +/* + * 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.pd.core.store; + +import org.rocksdb.Options; +import org.rocksdb.RocksDB; + +/** + * Holds a RocksDB LOCK from a separate JVM, the way a previous PD process does while it is + * still stopping. Prints {@link #LOCKED} once the database is open, then releases it after + * the given number of milliseconds, or holds it until killed when the number is negative. + */ +public final class RocksDBLockHolder { + + public static final String LOCKED = "LOCKED"; + + private RocksDBLockHolder() { + } + + public static void main(String[] args) throws Exception { + String dbPath = args[0]; + long holdMs = Long.parseLong(args[1]); + RocksDB.loadLibrary(); + try (Options options = new Options().setCreateIfMissing(true); + RocksDB ignored = RocksDB.open(options, dbPath)) { + System.out.println(LOCKED); + System.out.flush(); + Thread.sleep(holdMs < 0 ? Long.MAX_VALUE : holdMs); + } + } +}