You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by ji...@apache.org on 2026/10/08 15:34:59 UTC
(hugegraph) 09/15: fix(pd): drain work before closing metadata
This is an automated email from the ASF dual-hosted git repository.
imbajin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hugegraph.git
commit 97c9c779950881a2b3c624276b9247d711fb9ac0
Author: imbajin <ji...@apache.org>
AuthorDate: Thu Oct 8 20:31:40 2026 +0800
fix(pd): drain work before closing metadata
- coordinate scheduled services and Raft shutdown
- release metadata and native database owners
- cover PD lifecycle cleanup
---
.../apache/hugegraph/pd/TaskScheduleService.java | 16 ++
.../apache/hugegraph/pd/meta/MetadataFactory.java | 39 +++-
.../org/apache/hugegraph/pd/raft/RaftEngine.java | 90 +++++---
.../apache/hugegraph/pd/raft/RaftRpcClient.java | 7 +
.../apache/hugegraph/pd/raft/RaftStateMachine.java | 63 +++++-
.../apache/hugegraph/pd/store/HgKVStoreImpl.java | 50 +++--
.../org/apache/hugegraph/pd/boot/HugePDServer.java | 7 +-
.../org/apache/hugegraph/pd/boot/ShutdownHook.java | 139 ++++++++-----
.../apache/hugegraph/pd/pulse/PDPulseSubject.java | 21 ++
.../hugegraph/pd/service/KvServiceGrpcImpl.java | 27 +++
.../org/apache/hugegraph/pd/service/PDService.java | 17 ++
.../apache/hugegraph/pd/core/PDCoreSuiteTest.java | 1 +
.../hugegraph/pd/core/PDNativeOwnerTest.java | 226 +++++++++++++++++++++
13 files changed, 580 insertions(+), 123 deletions(-)
diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java
index 003a4278d..c111a6f85 100644
--- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java
+++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/TaskScheduleService.java
@@ -190,6 +190,22 @@ public class TaskScheduleService {
public void shutDown() {
executor.shutdownNow();
+ boolean interrupted = Thread.interrupted();
+ try {
+ while (!executor.isTerminated()) {
+ try {
+ if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
+ log.warn("Waiting for PD scheduled tasks to stop");
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
}
private boolean isLeader() {
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 673c313df..954a9f9e2 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
@@ -28,30 +28,47 @@ import org.apache.hugegraph.pd.store.RaftKVStore;
*/
public class MetadataFactory {
- private static HgKVStore store = null;
+ private static HgKVStore store;
+ private static boolean shutdown;
- public static HgKVStore getStore(PDConfig pdConfig) {
+ public static synchronized HgKVStore getStore(PDConfig pdConfig) {
+ if (shutdown) {
+ throw new IllegalStateException("PD metadata store has shut down");
+ }
if (store == null) {
- synchronized (MetadataFactory.class) {
- if (store == null) {
- HgKVStore proto = new HgKVStoreImpl();
- //proto.init(pdConfig);
- store = pdConfig.getRaft().isEnable() ?
- new RaftKVStore(RaftEngine.getInstance(), proto) :
- proto;
- store.init(pdConfig);
+ HgKVStore proto = new HgKVStoreImpl();
+ HgKVStore candidate = pdConfig.getRaft().isEnable() ?
+ new RaftKVStore(RaftEngine.getInstance(), proto) : proto;
+ try {
+ candidate.init(pdConfig);
+ store = candidate;
+ } catch (RuntimeException | Error failure) {
+ try {
+ candidate.close();
+ } catch (RuntimeException | Error cleanup) {
+ if (cleanup != failure) {
+ failure.addSuppressed(cleanup);
+ }
}
+ throw failure;
}
}
return store;
}
- public static void closeStore() {
+ public static synchronized void closeStore() {
if (store != null) {
store.close();
}
}
+ // Final service shutdown must not allow a late facade to create a new DB.
+ public static synchronized void shutdownStore() {
+ shutdown = true;
+ closeStore();
+ store = null;
+ }
+
public static StoreInfoMeta newStoreInfoMeta(PDConfig pdConfig) {
return new StoreInfoMeta(pdConfig);
}
diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java
index d39384909..50eb1ca90 100644
--- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java
+++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java
@@ -58,7 +58,6 @@ import com.alipay.sofa.jraft.rpc.RaftRpcServerFactory;
import com.alipay.sofa.jraft.rpc.RpcServer;
import com.alipay.sofa.jraft.rpc.impl.BoltRpcServer;
import com.alipay.sofa.jraft.util.Endpoint;
-import com.alipay.sofa.jraft.util.internal.ThrowUtil;
import io.netty.channel.ChannelHandler;
import lombok.extern.slf4j.Slf4j;
@@ -195,43 +194,68 @@ public class RaftEngine {
}
}
- public void shutDown() {
- if (this.alivePeersRefresher != null) {
- this.alivePeersRefresher.shutdownNow();
- try {
- // Best effort: shutdownNow only interrupts, and a refresh parked in
- // listAlivePeers waits on a lock acquire the interrupt does not break,
- // for up to the raft rpc connect timeout per unreachable peer. A refresh
- // that outlives this wait may publish one stale count over the reset
- // below; acceptable while shutDown has no production caller.
- if (!this.alivePeersRefresher.awaitTermination(1, TimeUnit.SECONDS)) {
- log.warn("Raft alive-peers refresher still running after shutdown; " +
- "hg_raft_alive_peers may briefly report a stale value");
+ // Call while Raft can still complete metadata work already issued by a listener.
+ public void stopLeaderCallbacks() {
+ this.stateMachine.stopLeaderCallbacks();
+ }
+
+ public synchronized void shutDown() {
+ boolean interrupted = Thread.interrupted();
+ try {
+ if (this.alivePeersRefresher != null) {
+ this.alivePeersRefresher.shutdownNow();
+ while (!this.alivePeersRefresher.isTerminated()) {
+ try {
+ if (!this.alivePeersRefresher.awaitTermination(5, TimeUnit.SECONDS)) {
+ log.warn("Waiting for PD Raft alive-peers refresher to stop");
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
}
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ this.alivePeersRefresher = null;
}
- this.alivePeersRefresher = null;
- }
- this.alivePeerCount = -1;
- if (this.raftGroupService != null) {
- this.raftGroupService.shutdown();
- try {
- this.raftGroupService.join();
- } catch (final InterruptedException e) {
+ this.alivePeerCount = -1;
+ if (this.raftGroupService != null) {
+ this.raftGroupService.shutdown();
+ boolean joined = false;
+ while (!joined) {
+ try {
+ this.raftGroupService.join();
+ joined = true;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ this.raftGroupService = null;
+ this.raftNode = null;
+ } else if (this.raftNode != null) {
+ this.raftNode.shutdown();
+ boolean joined = false;
+ while (!joined) {
+ try {
+ this.raftNode.join();
+ joined = true;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
this.raftNode = null;
- ThrowUtil.throwException(e);
}
- this.raftGroupService = null;
- }
- if (this.rpcServer != null) {
- this.rpcServer.shutdown();
- this.rpcServer = null;
- }
- if (this.raftNode != null) {
- this.raftNode.shutdown();
+ if (this.rpcServer != null) {
+ this.rpcServer.shutdown();
+ this.rpcServer = null;
+ }
+ if (this.raftRpcClient != null) {
+ this.raftRpcClient.shutdown();
+ this.raftRpcClient = null;
+ }
+ log.info("PD Raft shutdown joined");
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
- this.raftNode = null;
}
public boolean isLeader() {
diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java
index 8c7398a53..18602101f 100644
--- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java
+++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftRpcClient.java
@@ -45,6 +45,13 @@ public class RaftRpcClient {
return this.rpcClient.init(null);
}
+ public synchronized void shutdown() {
+ if (this.rpcClient != null) {
+ this.rpcClient.shutdown();
+ this.rpcClient = null;
+ }
+ }
+
/**
* Request a snapshot
*/
diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftStateMachine.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftStateMachine.java
index 65451297c..cc9caaa90 100644
--- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftStateMachine.java
+++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftStateMachine.java
@@ -80,6 +80,9 @@ public class RaftStateMachine extends StateMachineAdapter {
}
private List<RaftTaskHandler> taskHandlers;
private List<RaftStateListener> stateListeners;
+ private final Object callbackMonitor = new Object();
+ private int callbacks;
+ private boolean stoppingCallbacks;
public RaftStateMachine() {
this.taskHandlers = new CopyOnWriteArrayList<>();
@@ -150,6 +153,54 @@ public class RaftStateMachine extends StateMachineAdapter {
super.onShutdown();
}
+ private void notifyStateListeners() {
+ synchronized (this.callbackMonitor) {
+ if (this.stoppingCallbacks || CollectionUtils.isEmpty(this.stateListeners)) {
+ return;
+ }
+ this.callbacks++;
+ }
+ try {
+ Utils.runInThread(() -> {
+ try {
+ this.stateListeners.forEach(RaftStateListener::onRaftLeaderChanged);
+ } finally {
+ callbackFinished();
+ }
+ });
+ } catch (RuntimeException | Error failure) {
+ callbackFinished();
+ throw failure;
+ }
+ }
+
+ private void callbackFinished() {
+ synchronized (this.callbackMonitor) {
+ this.callbacks--;
+ this.callbackMonitor.notifyAll();
+ }
+ }
+
+ void stopLeaderCallbacks() {
+ boolean interrupted = false;
+ synchronized (this.callbackMonitor) {
+ this.stoppingCallbacks = true;
+ while (this.callbacks != 0) {
+ try {
+ this.callbackMonitor.wait(5000L);
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ if (this.callbacks != 0) {
+ log.warn("Waiting for PD leader callbacks to finish");
+ }
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
@Override
public void onLeaderStart(final long term) {
this.leaderTerm.set(term);
@@ -157,11 +208,7 @@ public class RaftStateMachine extends StateMachineAdapter {
super.onLeaderStart(term);
log.info("Raft becomes leader");
- Utils.runInThread(() -> {
- if (!CollectionUtils.isEmpty(stateListeners)) {
- stateListeners.forEach(RaftStateListener::onRaftLeaderChanged);
- }
- });
+ notifyStateListeners();
}
@Override
@@ -182,11 +229,7 @@ public class RaftStateMachine extends StateMachineAdapter {
public void onStartFollowing(final LeaderChangeContext ctx) {
this.probeView = new ProbeView(State.STATE_FOLLOWER, true);
super.onStartFollowing(ctx);
- Utils.runInThread(() -> {
- if (!CollectionUtils.isEmpty(stateListeners)) {
- stateListeners.forEach(RaftStateListener::onRaftLeaderChanged);
- }
- });
+ notifyStateListeners();
}
@Override
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 fd5b9c603..f33648e0e 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
@@ -62,12 +62,11 @@ public class HgKVStoreImpl implements HgKVStore {
@Override
public void init(PDConfig config) {
- dbOptions = new Options().setCreateIfMissing(true);
-
final Lock writeLock = this.readWriteLock.writeLock();
writeLock.lock();
try {
- this.dbPath = config.getDataPath() + "/rocksdb/";
+ this.dbOptions = new Options().setCreateIfMissing(true);
+ this.dbPath = config.getDataPath() + "/rocksdb";
File file = new File(this.dbPath);
if (!file.exists()) {
try {
@@ -77,13 +76,14 @@ public class HgKVStoreImpl implements HgKVStore {
}
}
openRocksDB(dbPath);
- } catch (PDException e) {
- // TODO: retry the open and then fail PD startup instead of logging: a held RocksDB LOCK
- // leaves this PD running uninitialized (/v1/ready answers 503 STATE_UNINITIALIZED
- // while /v1/health answers 200), and with several PDs nothing restarts it. The Helm
- // chart (helm/hugegraph) documents this as a limitation; drop that entry once fixed.
+ } catch (PDException | RuntimeException | Error e) {
+ // TODO: retry opening a held RocksDB LOCK before failing PD startup.
// https://github.com/apache/hugegraph/issues/3226
- log.error("Failed to open data file,{}", e);
+ if (this.dbOptions != null) {
+ this.dbOptions.close();
+ this.dbOptions = null;
+ }
+ throw new IllegalStateException("Failed to open PD metadata at " + this.dbPath, e);
} finally {
writeLock.unlock();
}
@@ -119,10 +119,10 @@ public class HgKVStoreImpl implements HgKVStore {
public List<KV> scanPrefix(byte[] prefix) {
final Lock readLock = this.readWriteLock.readLock();
readLock.lock();
- try (ReadOptions options = new ReadOptions()
- .setIterateLowerBound(new Slice(prefix))) {
+ try (Slice bound = new Slice(prefix);
+ ReadOptions options = new ReadOptions().setIterateLowerBound(bound);
+ RocksIterator iterator = db.newIterator(options)) {
List<KV> kvs = new ArrayList<>();
- RocksIterator iterator = db.newIterator(options);
iterator.seekToFirst();
while (iterator.isValid() && 0 == Bytes.indexOf(iterator.key(), prefix)) {
kvs.add(new KV(iterator.key(), iterator.value()));
@@ -152,9 +152,9 @@ public class HgKVStoreImpl implements HgKVStore {
public long removeByPrefix(byte[] prefix) throws PDException {
final Lock readLock = this.readWriteLock.readLock();
readLock.lock();
- try (ReadOptions options = new ReadOptions()
- .setIterateLowerBound(new Slice(prefix))) {
- RocksIterator iterator = db.newIterator(options);
+ try (Slice bound = new Slice(prefix);
+ ReadOptions options = new ReadOptions().setIterateLowerBound(bound);
+ RocksIterator iterator = db.newIterator(options)) {
iterator.seekToFirst();
while (iterator.isValid()) {
@@ -309,11 +309,11 @@ public class HgKVStoreImpl implements HgKVStore {
public List<KV> scanRange(byte[] start, byte[] end) {
final Lock readLock = this.readWriteLock.readLock();
readLock.lock();
- try (ReadOptions options = new ReadOptions()
- .setIterateLowerBound(new Slice(start))
- .setIterateUpperBound(new Slice(end))) {
+ try (Slice lower = new Slice(start);
+ Slice upper = new Slice(end);
+ ReadOptions options = new ReadOptions().setIterateLowerBound(lower).setIterateUpperBound(upper);
+ RocksIterator iterator = db.newIterator(options)) {
List<KV> kvs = new ArrayList<>();
- RocksIterator iterator = db.newIterator(options);
iterator.seekToFirst();
while (iterator.isValid()) {
kvs.add(new KV(iterator.key(), iterator.value()));
@@ -327,7 +327,17 @@ public class HgKVStoreImpl implements HgKVStore {
@Override
public void close() {
- closeRocksDB();
+ Lock writeLock = this.readWriteLock.writeLock();
+ writeLock.lock();
+ try {
+ closeRocksDB();
+ if (this.dbOptions != null) {
+ this.dbOptions.close();
+ this.dbOptions = null;
+ }
+ } finally {
+ writeLock.unlock();
+ }
}
private void closeRocksDB() {
diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/HugePDServer.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/HugePDServer.java
index 54d2761df..2ca7f7540 100644
--- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/HugePDServer.java
+++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/HugePDServer.java
@@ -17,6 +17,7 @@
package org.apache.hugegraph.pd.boot;
+import org.apache.hugegraph.util.RocksDBRuntime;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
@@ -37,8 +38,10 @@ public class HugePDServer {
System.setProperty("logging.path", "logs");
System.setProperty("com.alipay.remoting.client.log.level", "error");
}
- Runtime.getRuntime().addShutdownHook(new ShutdownHook(Thread.currentThread()));
- SpringApplication.run(HugePDServer.class);
+ SpringApplication application = new SpringApplication(HugePDServer.class);
+ application.addInitializers(context -> RocksDBRuntime.verify(
+ context.getEnvironment().getProperty("rocksdb.provider", "rocksdb")));
+ application.run(args);
System.out.println("Hugegraph-pd started.");
}
}
diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/ShutdownHook.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/ShutdownHook.java
index 371235f34..a8cf28abb 100644
--- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/ShutdownHook.java
+++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/boot/ShutdownHook.java
@@ -15,75 +15,120 @@
* limitations under the License.
*/
-
package org.apache.hugegraph.pd.boot;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.pd.TaskScheduleService;
+import org.apache.hugegraph.pd.meta.MetadataFactory;
+import org.apache.hugegraph.pd.raft.RaftEngine;
+import org.apache.hugegraph.pd.pulse.PDPulseSubject;
+import org.apache.hugegraph.pd.service.KvServiceGrpcImpl;
import org.apache.hugegraph.pd.service.MetadataService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.hugegraph.pd.service.PDService;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.ApplicationListener;
+import org.lognet.springboot.grpc.context.GRpcServerInitializedEvent;
-import java.util.concurrent.ThreadPoolExecutor;
+import io.grpc.Server;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
-public class ShutdownHook extends Thread {
+/**
+ * Spring stops its transports before destroying singleton beans. The PDService
+ * dependency keeps its metadata owners available until this cleanup completes.
+ */
+@Slf4j
+@Component
+public class ShutdownHook implements DisposableBean, ApplicationListener<GRpcServerInitializedEvent> {
- private static Logger log = LoggerFactory.getLogger(ShutdownHook.class);
- private static String msg = "there are still uninterruptible jobs that have not been completed and" +
- " will wait for them to complete";
- private Thread main;
+ private final PDService service;
+ private final KvServiceGrpcImpl kvService;
+ private Server grpcServer;
- public ShutdownHook(Thread main) {
- super();
- this.main = main;
+ public ShutdownHook(PDService service, KvServiceGrpcImpl kvService) {
+ this.service = service;
+ this.kvService = kvService;
}
@Override
- public void run() {
- log.info("shutdown signal received");
- main.interrupt();
- waitForShutdown();
- try {
- main.join();
- } catch (InterruptedException e) {
- }
- log.info("shutdown completed");
+ public void onApplicationEvent(GRpcServerInitializedEvent event) {
+ this.grpcServer = event.getServer();
}
- private void waitForShutdown() {
- checkUninterruptibleJobs();
+ @Override
+ public void destroy() {
+ // The starter's default stop initiates shutdown without waiting.
+ awaitGrpcTermination();
+ this.kvService.stopScheduling();
+ closeOwners(this.service);
}
- private void checkUninterruptibleJobs() {
- ThreadPoolExecutor jobs = MetadataService.getUninterruptibleJobs();
+ public static void closeOwners(PDService service) {
+ log.info("Closing PD background work and native owners");
+ TaskScheduleService tasks = service.getTaskService();
+ if (tasks != null) {
+ tasks.shutDown();
+ }
+ PDPulseSubject.stopScheduling();
+ RaftEngine raft = RaftEngine.getInstance();
+ raft.stopLeaderCallbacks();
+ // Raft may still submit snapshot saves while its shutdown is joining.
+ raft.shutDown();
+ awaitUninterruptibleJobs();
+ // A failed join must not release the database out from under Raft.
+ MetadataFactory.shutdownStore();
+ log.info("PD native owners closed");
+ }
+
+ private void awaitGrpcTermination() {
+ if (this.grpcServer == null) {
+ return;
+ }
+ // End long-lived watch/pulse streams as well as accepting no new calls.
+ this.grpcServer.shutdownNow();
+ boolean interrupted = Thread.interrupted();
try {
- if (jobs != null) {
- long lastPrint = System.currentTimeMillis() - 5000;
- log.info("check for ongoing background jobs that cannot be interrupted, active:{}, queue:{}.",
- jobs.getActiveCount(), jobs.getQueue().size());
- while (jobs.getActiveCount() != 0 || jobs.getQueue().size() != 0) {
- synchronized (ShutdownHook.class) {
- if (System.currentTimeMillis() - lastPrint > 5000) {
- log.warn(msg);
- lastPrint = System.currentTimeMillis();
- }
- try {
- ShutdownHook.class.wait(200);
- } catch (InterruptedException e) {
- log.error("close jobs with error:", e);
- }
+ while (!this.grpcServer.isTerminated()) {
+ try {
+ if (!this.grpcServer.awaitTermination(5, TimeUnit.SECONDS)) {
+ log.warn("Waiting for PD gRPC transport to terminate");
}
+ } catch (InterruptedException e) {
+ interrupted = true;
}
- log.info("all ongoing background jobs have been completed and the shutdown will continue");
}
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
- } catch (Exception e) {
- log.error("close jobs with error:", e);
+ private static void awaitUninterruptibleJobs() {
+ ThreadPoolExecutor jobs = MetadataService.getUninterruptibleJobs();
+ if (jobs == null) {
+ return;
}
+ jobs.shutdown();
+ boolean interrupted = Thread.interrupted();
try {
- if (jobs != null) {
- jobs.shutdownNow();
+ while (!jobs.isTerminated()) {
+ try {
+ if (!jobs.awaitTermination(5, TimeUnit.SECONDS)) {
+ log.warn("Waiting for PD metadata jobs, active: {}, queued: {}",
+ jobs.getActiveCount(), jobs.getQueue().size());
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
}
- } catch (Exception e) {
- log.error("close jobs with error:", e);
}
}
}
diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/pulse/PDPulseSubject.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/pulse/PDPulseSubject.java
index 86b95cb88..58d886abe 100644
--- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/pulse/PDPulseSubject.java
+++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/pulse/PDPulseSubject.java
@@ -89,6 +89,27 @@ public class PDPulseSubject {
TimeUnit.SECONDS);
}
+ /** Stop periodic queue work and wait while Raft can still serve its writes. */
+ public static void stopScheduling() {
+ scheduledExecutor.shutdown();
+ boolean interrupted = Thread.interrupted();
+ try {
+ while (!scheduledExecutor.isTerminated()) {
+ try {
+ if (!scheduledExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
+ log.warn("Waiting for PD pulse scheduler to stop");
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
private static void doSchedule() {
appendQueue();
expireQueue();
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 e49a4dc48..94b41a0f0 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
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
import org.apache.hugegraph.pd.KvService;
import org.apache.hugegraph.pd.common.PDException;
@@ -86,6 +87,32 @@ public class KvServiceGrpcImpl extends KvServiceGrpc.KvServiceImplBase implement
}, 0, KvWatchSubject.WATCH_TTL * 1 / 3, TimeUnit.MILLISECONDS);
}
+ /** Stop periodic watch work and wait while Raft can still serve its writes. */
+ @PreDestroy
+ public void stopScheduling() {
+ ScheduledExecutorService current = this.executor;
+ if (current == null) {
+ return;
+ }
+ current.shutdown();
+ boolean interrupted = Thread.interrupted();
+ try {
+ while (!current.isTerminated()) {
+ try {
+ if (!current.awaitTermination(5, TimeUnit.SECONDS)) {
+ log.warn("Waiting for PD KV watch scheduler to stop");
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
/**
* Ordinary put
*
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 94d136a84..f40b2f580 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
@@ -53,6 +53,7 @@ import org.apache.hugegraph.pd.StoreMonitorDataService;
import org.apache.hugegraph.pd.StoreNodeService;
import org.apache.hugegraph.pd.StoreStatusListener;
import org.apache.hugegraph.pd.TaskScheduleService;
+import org.apache.hugegraph.pd.boot.ShutdownHook;
import org.apache.hugegraph.pd.common.KVPair;
import org.apache.hugegraph.pd.common.PDException;
import org.apache.hugegraph.pd.config.PDConfig;
@@ -177,6 +178,22 @@ public class PDService extends PDGrpc.PDImplBase implements RaftStateListener {
*/
@PostConstruct
public void init() throws PDException {
+ try {
+ initService();
+ } catch (PDException | RuntimeException | Error failure) {
+ // Failed @PostConstruct beans are not registered for destruction.
+ try {
+ ShutdownHook.closeOwners(this);
+ } catch (RuntimeException | Error cleanup) {
+ if (cleanup != failure) {
+ failure.addSuppressed(cleanup);
+ }
+ }
+ throw failure;
+ }
+ }
+
+ private void initService() throws PDException {
log.info("PDService init………… {}", pdConfig);
configService = new ConfigService(pdConfig);
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 ddc1fad7f..1c7317670 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
@@ -33,6 +33,7 @@ import lombok.extern.slf4j.Slf4j;
RaftStateMachineTest.class,
MetadataKeyHelperTest.class,
HgKVStoreImplTest.class,
+ PDNativeOwnerTest.class,
PDConfigTest.class,
ConfigServiceTest.class,
IdServiceTest.class,
diff --git a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/PDNativeOwnerTest.java b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/PDNativeOwnerTest.java
new file mode 100644
index 000000000..e7c6c1d05
--- /dev/null
+++ b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/PDNativeOwnerTest.java
@@ -0,0 +1,226 @@
+/*
+ * 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 java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.nio.file.Path;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReadWriteLock;
+
+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.raft.RaftRpcClient;
+import org.apache.hugegraph.pd.raft.RaftStateMachine;
+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.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDB;
+
+import com.alipay.sofa.jraft.Node;
+import com.alipay.sofa.jraft.RaftGroupService;
+import com.alipay.sofa.jraft.rpc.RpcServer;
+
+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;
+
+public class PDNativeOwnerTest {
+
+ @Rule
+ public TemporaryFolder temporary = new TemporaryFolder();
+
+ @Test(timeout = 10000)
+ public void testCloseWaitsForReadersAndReleasesNativeOwners() throws Exception {
+ RocksDB.loadLibrary();
+ Path path = this.temporary.newFolder().toPath();
+ PDConfig config = new PDConfig();
+ config.setDataPath(path.toString());
+ HgKVStoreImpl store = new HgKVStoreImpl();
+ store.init(config);
+ RocksDB db = Whitebox.getInternalState(store, "db");
+ Options options = Whitebox.getInternalState(store, "dbOptions");
+ ReadWriteLock lock = Whitebox.getInternalState(store, "readWriteLock");
+ store.put(new byte[]{1}, new byte[]{2});
+ store.put(new byte[]{3}, new byte[]{4});
+ Assert.assertEquals(1, store.scanPrefix(new byte[]{1}).size());
+ Assert.assertEquals(2, store.scanRange(new byte[]{0}, new byte[]{4}).size());
+ store.removeByPrefix(new byte[]{1});
+ Assert.assertEquals(0, store.scanPrefix(new byte[]{1}).size());
+ CountDownLatch closing = new CountDownLatch(1);
+ FutureTask<Void> close = new FutureTask<>(() -> {
+ closing.countDown();
+ store.close();
+ return null;
+ });
+ Thread closer = new Thread(close, "pd-owner-close");
+ lock.readLock().lock();
+ try {
+ closer.start();
+ Assert.assertTrue(closing.await(1, TimeUnit.SECONDS));
+ Assert.assertThrows(TimeoutException.class, () -> close.get(50, TimeUnit.MILLISECONDS));
+ Assert.assertTrue(db.isOwningHandle());
+ Assert.assertTrue(options.isOwningHandle());
+ } finally {
+ lock.readLock().unlock();
+ }
+ close.get(2, TimeUnit.SECONDS);
+ store.close();
+ Assert.assertFalse(db.isOwningHandle());
+ Assert.assertFalse(options.isOwningHandle());
+ HgKVStoreImpl reopened = new HgKVStoreImpl();
+ reopened.init(config);
+ try {
+ Assert.assertArrayEquals(new byte[]{4}, reopened.get(new byte[]{3}));
+ } finally {
+ reopened.close();
+ }
+ }
+
+ @Test
+ public void testInitializationFailureReleasesOptions() {
+ RocksDB.loadLibrary();
+ HgKVStoreImpl store = new HgKVStoreImpl();
+ PDConfig config = mock(PDConfig.class);
+ AtomicReference<Options> allocated = new AtomicReference<>();
+ IllegalStateException injected = new IllegalStateException("injected configuration failure");
+ when(config.getDataPath()).thenAnswer(invocation -> {
+ allocated.set(Whitebox.getInternalState(store, "dbOptions"));
+ throw injected;
+ });
+ IllegalStateException failure = Assert.assertThrows(IllegalStateException.class, () -> store.init(config));
+ Assert.assertSame(injected, failure.getCause());
+ Assert.assertFalse(allocated.get().isOwningHandle());
+ store.close();
+ }
+
+ @Test
+ public void testTerminalFactoryCloseRejectsLazyReopen() throws Exception {
+ Field storeField = MetadataFactory.class.getDeclaredField("store");
+ Field shutdownField = MetadataFactory.class.getDeclaredField("shutdown");
+ storeField.setAccessible(true);
+ shutdownField.setAccessible(true);
+ synchronized (MetadataFactory.class) {
+ Object previous = storeField.get(null);
+ boolean shutdown = shutdownField.getBoolean(null);
+ HgKVStore owned = mock(HgKVStore.class);
+ try {
+ storeField.set(null, owned);
+ shutdownField.setBoolean(null, false);
+ MetadataFactory.shutdownStore();
+ MetadataFactory.shutdownStore();
+ verify(owned, times(1)).close();
+ Assert.assertThrows(IllegalStateException.class, () -> MetadataFactory.getStore(null));
+ } finally {
+ storeField.set(null, previous);
+ shutdownField.setBoolean(null, shutdown);
+ }
+ }
+ }
+
+ @Test
+ public void testInterruptedRaftJoinRetainsOwnerAndClosesRpcClient() throws Exception {
+ RaftEngine engine = newEngine();
+ RaftGroupService group = mock(RaftGroupService.class);
+ Node node = mock(Node.class);
+ RpcServer server = mock(RpcServer.class);
+ RaftRpcClient client = mock(RaftRpcClient.class);
+ Whitebox.setInternalState(engine, "raftGroupService", group);
+ Whitebox.setInternalState(engine, "raftNode", node);
+ Whitebox.setInternalState(engine, "rpcServer", server);
+ Whitebox.setInternalState(engine, "raftRpcClient", client);
+ AtomicInteger attempts = new AtomicInteger();
+ doAnswer(invocation -> {
+ Assert.assertSame(node, engine.getRaftNode());
+ if (attempts.incrementAndGet() == 1) {
+ throw new InterruptedException("injected join interruption");
+ }
+ return null;
+ }).when(group).join();
+ try {
+ engine.shutDown();
+ Assert.assertTrue(Thread.interrupted());
+ Assert.assertEquals(2, attempts.get());
+ Assert.assertNull(engine.getRaftNode());
+ engine.shutDown();
+ verify(group, times(1)).shutdown();
+ verify(node, never()).shutdown();
+ verify(server, times(1)).shutdown();
+ verify(client, times(1)).shutdown();
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ @Test(timeout = 10000)
+ public void testLeaderCallbacksDrainBeforeStoppingRaft() throws Exception {
+ RaftEngine engine = newEngine();
+ RaftStateMachine state = Whitebox.getInternalState(engine, "stateMachine");
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger calls = new AtomicInteger();
+ state.addStateListener(() -> {
+ calls.incrementAndGet();
+ entered.countDown();
+ try {
+ Assert.assertTrue(release.await(3, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ throw new AssertionError(e);
+ }
+ });
+ state.onLeaderStart(1);
+ Assert.assertTrue(entered.await(1, TimeUnit.SECONDS));
+ FutureTask<Void> stop = new FutureTask<>(() -> {
+ engine.stopLeaderCallbacks();
+ return null;
+ });
+ Thread stopper = new Thread(stop, "pd-leader-drain");
+ try {
+ stopper.start();
+ Assert.assertThrows(TimeoutException.class, () -> stop.get(50, TimeUnit.MILLISECONDS));
+ release.countDown();
+ stop.get(2, TimeUnit.SECONDS);
+ state.onLeaderStart(2);
+ engine.stopLeaderCallbacks();
+ Assert.assertEquals(1, calls.get());
+ } finally {
+ release.countDown();
+ stopper.join(3000);
+ }
+ }
+
+ private static RaftEngine newEngine() throws Exception {
+ Constructor<RaftEngine> constructor = RaftEngine.class.getDeclaredConstructor();
+ constructor.setAccessible(true);
+ return constructor.newInstance();
+ }
+}