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/16 14:05:57 UTC
(hugegraph) branch master updated: refactor(server): coordinate graph rpc lifecycle (#3329)
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
The following commit(s) were added to refs/heads/master by this push:
new 6f98193ea refactor(server): coordinate graph rpc lifecycle (#3329)
6f98193ea is described below
commit 6f98193eac966421e499f5d506d86b22094b52cd
Author: imbajin <ji...@apache.org>
AuthorDate: Fri Oct 16 22:05:52 2026 +0800
refactor(server): coordinate graph rpc lifecycle (#3329)
Graphs added after startup could miss cache RPC registration. Cleanup failures could retain stale owners, and a same-name ADD could be discarded while an older owner was still healthy or closing. Graphs now become visible after RPC registration, unfinished cleanup remains retryable, and received ADDs reload current metadata after the old owner is released.
---
docs/server-runtime.md | 81 +
.../hugegraph/rpc/RpcServiceConfig4Client.java | 4 +
.../hugegraph/rpc/RpcServiceConfig4Server.java | 4 +
.../apache/hugegraph/rpc/RpcConsumerConfig.java | 131 +-
.../apache/hugegraph/rpc/RpcProviderConfig.java | 54 +-
.../java/org/apache/hugegraph/rpc/RpcServer.java | 6 +-
.../org/apache/hugegraph/unit/BaseUnitTest.java | 37 +-
.../org/apache/hugegraph/unit/RpcConfigTest.java | 297 +-
.../apache/hugegraph/unit/ServerClientTest.java | 220 +-
hugegraph-server/README.md | 2 +
.../apache/hugegraph/auth/HugeGraphAuthProxy.java | 6 +
.../org/apache/hugegraph/core/GraphManager.java | 445 ++-
.../main/java/org/apache/hugegraph/HugeGraph.java | 5 +
.../java/org/apache/hugegraph/LoadingMode.java | 4 +-
.../org/apache/hugegraph/StandardHugeGraph.java | 201 +-
.../store/raft/RaftBackendStoreProvider.java | 31 +-
.../hugegraph/backend/store/raft/RaftContext.java | 76 +-
.../hugegraph/backend/store/raft/RaftNode.java | 16 +-
.../hugegraph/task/DistributedTaskScheduler.java | 12 +-
.../hugegraph/task/StandardTaskScheduler.java | 25 +-
.../org/apache/hugegraph/util/CloseFailure.java | 35 +
.../hugegraph/DistributedTaskAdmissionTest.java | 58 +
.../hugegraph/StandardTaskAdmissionTest.java | 485 +++
.../unit/core/GraphManagerConfigTest.java | 3367 ++++++++++++++++++++
24 files changed, 5344 insertions(+), 258 deletions(-)
diff --git a/docs/server-runtime.md b/docs/server-runtime.md
index 586d2d11c..d3466cce1 100644
--- a/docs/server-runtime.md
+++ b/docs/server-runtime.md
@@ -481,6 +481,9 @@ Task database workers release their owned task and graph transactions on the wor
thread. A failure while closing one owner must not skip cleanup of other owners or
server-info resources. Cleanup retains the first exception and attaches later cleanup
failures as suppressed exceptions.
+For the local Standard scheduler, a deleted task disappears from public lookup after its persisted record is
+removed, while its actual execution still counts toward drain until native cleanup finishes; late completion saves from
+that owner do not recreate the removed record.
Synchronous task cancellation preserves a transaction already open on its caller,
including uncommitted graph writes. Cancellation still runs the normal completion and
@@ -555,6 +558,84 @@ compatibility evidence. Run the read-only application examples against represent
existing data and validate the serializers, client versions and authentication
combinations actually used by the application.
+## Cache RPC graph lifecycle
+
+When cache RPC is enabled and a backend declares shared storage, Server registers both
+graph and schema cache notifiers for graphs loaded during startup and graphs created or
+loaded later. A new graph's RPC registration must succeed before its local publication.
+Graph services use the graph space and graph name together as their service identity.
+PD deletion and startup coordinate RPC removal for the same graph instance. Startup does not
+register an owner whose RPC removal is pending, even after its placement-event token is consumed,
+and checks logical closure under the actual owner's RPC lifecycle monitor before registration.
+Any retained provider configuration must be unregistered before startup can export services;
+a cleanup failure fails startup instead of re-exporting that owner. Graph clearing, task drains
+and full graph close remain outside this RPC publication gate.
+
+Internal graph loading registers RPC services in Server's administrative scope and
+restores the caller's identity afterward. Direct registration and cleanup remain
+administrative operations.
+
+Successful graph close releases that graph's notifier listeners, providers and consumer
+proxies. A failed task drain during ordinary close or non-clearing local removal retains
+the owner and its RPC resources. In PD-enabled deployments, RPC cleanup failures during
+graph DELETE retain the local owner and bindings for retry; partial cleanup does not imply
+a healthy service. This DELETE path first closes task admission and drains local executions,
+then cleans up RPC before metadata deletion and backend clearing. A failed drain or RPC
+cleanup can be retried with the same DELETE. Logical graph closure fences access before
+all resource cleanup completes. Any close failure retains the local owner and bindings;
+retrying the same DELETE retries cleanup without repeating backend clearing. A completed clear
+is remembered for that exact graph instance even if an external idle transaction delays close;
+retry after that transaction exits. A replacement graph with the same name is cleared independently.
+A graph closed by a non-clearing removal has no successful-clear certificate. A later clearing DELETE
+refuses that closed owner and keeps its binding, rather than reporting success while old backend data remains.
+Raft provider, node and an owned listener are closed independently. The first fatal Java Error takes
+priority over ordinary close failures; otherwise the first failure is retained. Other failures are
+reported as suppressed, with a repeated reference to the primary failure ignored.
+A graph using Server's shared Raft listener closes
+only its own node. Server releases that listener and its address registration on shutdown;
+deleting one graph keeps the listener available to other local graphs. An interrupted node
+join retains the graph owner and interrupt status so a later DELETE can finish cleanup.
+The context's backend, snapshot and optional read-index executors stop after node join and
+drain queued work before close completes. Their wait uses `task.wait_timeout`; timeout or
+interruption retains the graph for retry without cancelling queued operations.
+RPC cleanup can start before graph close finishes. Clearing PD DELETE unregisters RPC before
+metadata removal and backend clear. Factory, binding and quota removal require a successful graph close;
+a failed close retains their owner even if its RPC services have already been unregistered.
+One failing graph does not prevent other graphs in a PD removal batch from being processed.
+Fatal Java Errors retain their original identity and
+propagate out of the removal batch. Failures after backend clearing do not undo deletion;
+retries do not restore deleted data or placement metadata.
+
+For local deletion without PD, backend clearing precedes RPC cleanup during graph close.
+A close failure retains the same local graph, factory identity and configuration for DELETE retry,
+even when access has already been fenced by logical closure. Only a successful clear by that
+instance's local drop permits retry to skip clearing and finish its retained resource cleanup.
+An ordinary close does not provide that certificate. A failure before clearing keeps the open
+owner's existing RPC services. A later cleanup failure does not preserve or restore deleted data.
+
+Failed creation or local loading closes the candidate in Server's administrative scope.
+If rollback close fails, Server retains the local owner and factory identity for cleanup retry.
+Retained failed candidates are excluded from RPC startup even while they are still open.
+An old graph's cleanup cannot remove or re-export a replacement graph's services.
+A PD graph-add notification blocked by an existing owner is coalesced until that owner is
+released, including when independent watches deliver the new ADD before the old REMOVE.
+Duplicate ADD notifications alone keep the healthy owner. Server consumes the PD acknowledgement
+of its own local graph creation without retaining a replacement ADD unless that exact owner already
+has failed peer cleanup; the existing cleanup retry takes priority. A healthy creation acknowledgement
+cannot republish the graph during a later reload. After removal, Server rereads the latest configuration;
+a missing configuration creates nothing. Reload consumes pending ADDs only after each old owner is
+successfully removed, then loads current metadata together instead of replaying ADDs during removal.
+After a peer removal fails, a later same-name PD add retries non-clearing cleanup of that exact
+owner before loading its replacement. Another cleanup failure keeps the owner for the next add
+or normal reload; no background retry is scheduled. A replacement load failure is logged and
+remains pending for a later add or normal reload. The completed old-owner DELETE still succeeds.
+Events queued for a closed notifier are ignored; an RPC request already sent cannot be recalled.
+
+Cache RPC uses the configured peer addresses; local provider export does not establish
+that every remote consumer connection is ready. Notifications are asynchronous, and
+connection or delivery failures are logged. This lifecycle handling does not provide
+acknowledged delivery or a distributed cache consistency protocol. Validate actual DML
+and cache behavior against the deployment's shared backend and topology.
## Deleting a graph with local tasks
diff --git a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Client.java b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Client.java
index b08d6327d..e0557ebc2 100644
--- a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Client.java
+++ b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Client.java
@@ -32,4 +32,8 @@ public interface RpcServiceConfig4Client {
}
void removeAllServiceProxy();
+
+ default boolean removeServiceProxy(String graph, String interfaceId, Object proxy) {
+ throw new UnsupportedOperationException("Removing an individual RPC proxy is not supported");
+ }
}
diff --git a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Server.java b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Server.java
index cd0290900..0c9eab15a 100644
--- a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Server.java
+++ b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rpc/RpcServiceConfig4Server.java
@@ -25,5 +25,9 @@ public interface RpcServiceConfig4Server {
void removeService(String serviceId);
+ default boolean removeService(String serviceId, Object serviceImpl) {
+ throw new UnsupportedOperationException("Removing an RPC service by implementation is not supported");
+ }
+
void removeAllService();
}
diff --git a/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcConsumerConfig.java b/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcConsumerConfig.java
index 9459ac516..96819072e 100644
--- a/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcConsumerConfig.java
+++ b/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcConsumerConfig.java
@@ -20,7 +20,9 @@ package org.apache.hugegraph.rpc;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.RpcOptions;
@@ -39,9 +41,6 @@ import com.alipay.sofa.rpc.core.request.SofaRequest;
import com.alipay.sofa.rpc.core.response.SofaResponse;
import com.alipay.sofa.rpc.ext.Extension;
import com.alipay.sofa.rpc.ext.ExtensionLoaderFactory;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
public class RpcConsumerConfig implements RpcServiceConfig4Client {
@@ -59,8 +58,8 @@ public class RpcConsumerConfig implements RpcServiceConfig4Client {
RpcCommonConfig.initRpcConfigs(config);
this.conf = config;
this.remoteUrls = remoteUrls;
- this.configs = Maps.newHashMap();
- this.bootstraps = Lists.newArrayList();
+ this.configs = new ConcurrentHashMap<>();
+ this.bootstraps = new CopyOnWriteArrayList<>();
}
@Override
@@ -70,28 +69,119 @@ public class RpcConsumerConfig implements RpcServiceConfig4Client {
@Override
public <T> T serviceProxy(String graph, String interfaceId) {
- ConsumerConfig<T> config = this.consumerConfig(graph, interfaceId);
- ConsumerBootstrap<T> bootstrap = Bootstraps.from(config);
- this.bootstraps.add(bootstrap);
- return bootstrap.refer();
+ String serviceId = graph == null ? interfaceId : interfaceId + ":" + graph;
+ AtomicReference<T> proxy = new AtomicReference<>();
+ this.configs.compute(serviceId, (key, existing) -> {
+ @SuppressWarnings("unchecked")
+ ConsumerConfig<T> config = existing == null ? this.consumerConfig(graph, interfaceId) :
+ (ConsumerConfig<T>) existing;
+ ConsumerBootstrap<T> bootstrap = Bootstraps.from(config);
+ try {
+ proxy.set(bootstrap.refer());
+ this.bootstraps.add(bootstrap);
+ } catch (RuntimeException | Error failure) {
+ // Retain the cleanup owner even when refer never returned a proxy.
+ this.bootstraps.add(bootstrap);
+ throwFailure(this.cleanup(bootstrap, failure));
+ }
+ return config;
+ });
+ return proxy.get();
+ }
+
+ @Override
+ public boolean removeServiceProxy(String graph, String interfaceId, Object proxy) {
+ String serviceId = graph == null ? interfaceId : interfaceId + ":" + graph;
+ boolean[] removed = {false};
+ this.configs.computeIfPresent(serviceId, (key, config) -> {
+ for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
+ if (bootstrap.getConsumerConfig() == config && bootstrap.getProxyIns() == proxy) {
+ Throwable failure = this.cleanup(bootstrap, null);
+ if (failure != null) {
+ throwFailure(failure);
+ }
+ removed[0] = true;
+ break;
+ }
+ }
+ for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
+ if (bootstrap.getConsumerConfig() == config) {
+ return config;
+ }
+ }
+ return null;
+ });
+ return removed[0];
}
@Override
public void removeAllServiceProxy() {
+ Throwable failure = null;
for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
- bootstrap.unRefer();
+ try {
+ bootstrap.unRefer();
+ } catch (RuntimeException | Error cleanup) {
+ failure = addFailure(failure, cleanup);
+ }
+ }
+ if (failure != null) {
+ throwFailure(failure);
}
}
public void destroy() {
- Set<Cluster> clusters = Sets.newHashSet();
+ Throwable failure = null;
for (ConsumerBootstrap<?> bootstrap : this.bootstraps) {
+ failure = this.cleanup(bootstrap, failure);
+ }
+ if (failure != null) {
+ throwFailure(failure);
+ }
+ }
+
+ private Throwable cleanup(ConsumerBootstrap<?> bootstrap, Throwable failure) {
+ boolean complete = true;
+ try {
bootstrap.unRefer();
- clusters.add(bootstrap.getCluster());
+ } catch (RuntimeException | Error cleanup) {
+ complete = false;
+ failure = addFailure(failure, cleanup);
+ }
+ try {
+ Cluster cluster = bootstrap.getCluster();
+ if (cluster != null) {
+ cluster.destroy();
+ }
+ } catch (RuntimeException | Error cleanup) {
+ complete = false;
+ failure = addFailure(failure, cleanup);
+ }
+ if (complete) {
+ this.bootstraps.remove(bootstrap);
+ }
+ return failure;
+ }
+
+ private static Throwable addFailure(Throwable failure, Throwable cleanup) {
+ if (failure == null) {
+ return cleanup;
+ }
+ if (failure == cleanup) {
+ return failure;
}
- for (Cluster cluster : clusters) {
- cluster.destroy();
+ if (cleanup instanceof Error && !(failure instanceof Error)) {
+ cleanup.addSuppressed(failure);
+ return cleanup;
}
+ failure.addSuppressed(cleanup);
+ return failure;
+ }
+
+ private static void throwFailure(Throwable failure) {
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw (RuntimeException) failure;
}
private <T> ConsumerConfig<T> consumerConfig(String graph,
@@ -103,15 +193,7 @@ public class RpcConsumerConfig implements RpcServiceConfig4Client {
serviceId = interfaceId;
}
- @SuppressWarnings("unchecked")
- ConsumerConfig<T> consumerConfig = (ConsumerConfig<T>)
- this.configs.get(serviceId);
- if (consumerConfig != null) {
- return consumerConfig;
- }
-
- assert consumerConfig == null;
- consumerConfig = new ConsumerConfig<>();
+ ConsumerConfig<T> consumerConfig = new ConsumerConfig<>();
HugeConfig conf = this.conf;
String protocol = conf.get(RpcOptions.RPC_PROTOCOL);
@@ -138,7 +220,6 @@ public class RpcConsumerConfig implements RpcServiceConfig4Client {
.setSerialization(conf.get(RpcOptions.RPC_SERIALIZATION))
.setLoadBalancer(loadBalancer);
- this.configs.put(serviceId, consumerConfig);
return consumerConfig;
}
diff --git a/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcProviderConfig.java b/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcProviderConfig.java
index dafdfcf5a..3ad955a86 100644
--- a/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcProviderConfig.java
+++ b/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcProviderConfig.java
@@ -18,14 +18,15 @@
package org.apache.hugegraph.rpc;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.config.ServerConfig;
import org.apache.hugegraph.util.E;
-import com.google.common.collect.Maps;
public class RpcProviderConfig implements RpcServiceConfig4Server {
- private final Map<String, ProviderConfig<?>> configs = Maps.newHashMap();
+ private final Map<String, ProviderConfig<?>> configs = new ConcurrentHashMap<>();
@Override
public <T, S extends T> String addService(Class<T> clazz, S serviceImpl) {
@@ -54,27 +55,58 @@ public class RpcProviderConfig implements RpcServiceConfig4Server {
providerConfig.setInterfaceId(interfaceId)
.setRef(serviceImpl);
- E.checkArgument(!this.configs.containsKey(serviceId),
+ E.checkArgument(this.configs.putIfAbsent(serviceId, providerConfig) == null,
"Not allowed to add service already exist: '%s'",
serviceId);
- this.configs.put(serviceId, providerConfig);
return serviceId;
}
@Override
public void removeService(String serviceId) {
- ProviderConfig<?> config = this.configs.remove(serviceId);
- E.checkArgument(config != null,
- "The service '%s' doesn't exist", serviceId);
- config.unExport();
+ boolean[] removed = {false};
+ this.configs.computeIfPresent(serviceId, (key, config) -> {
+ config.unExport();
+ removed[0] = true;
+ return null;
+ });
+ E.checkArgument(removed[0], "The service '%s' doesn't exist", serviceId);
}
@Override
- public void removeAllService() {
- for (ProviderConfig<?> config : this.configs.values()) {
+ public boolean removeService(String serviceId, Object serviceImpl) {
+ boolean[] removed = {false};
+ this.configs.computeIfPresent(serviceId, (key, config) -> {
+ if (config.getRef() != serviceImpl) {
+ return config;
+ }
config.unExport();
+ removed[0] = true;
+ return null;
+ });
+ return removed[0];
+ }
+
+ @Override
+ public void removeAllService() {
+ for (String serviceId : this.configs.keySet()) {
+ this.configs.computeIfPresent(serviceId, (key, config) -> {
+ config.unExport();
+ return null;
+ });
+ }
+ }
+
+ void exportAll(ServerConfig server, int timeout) {
+ for (String serviceId : this.configs.keySet()) {
+ this.configs.computeIfPresent(serviceId, (key, config) -> {
+ // Sofa appends a single server; retain other bindings without adding duplicates.
+ if (config.getServer() == null || config.getServer().stream().noneMatch(bound -> bound == server)) {
+ config.setServer(server);
+ }
+ config.setTimeout(timeout).export();
+ return config;
+ });
}
- this.configs.clear();
}
public Map<String, ProviderConfig<?>> configs() {
diff --git a/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcServer.java b/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcServer.java
index 7791ac8cd..1af841c1c 100644
--- a/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcServer.java
+++ b/hugegraph-commons/hugegraph-rpc/src/main/java/org/apache/hugegraph/rpc/RpcServer.java
@@ -104,11 +104,7 @@ public class RpcServer {
return;
}
int timeout = this.conf.get(RpcOptions.RPC_SERVER_TIMEOUT) * 1000;
- for (ProviderConfig<?> providerConfig : configs.values()) {
- providerConfig.setServer(this.serverConfig)
- .setTimeout(timeout)
- .export();
- }
+ this.configs.exportAll(this.serverConfig, timeout);
LOG.info("RpcServer started success on port {}", this.port());
}
diff --git a/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/BaseUnitTest.java b/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/BaseUnitTest.java
index 34912bc76..5098f3f2a 100644
--- a/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/BaseUnitTest.java
+++ b/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/BaseUnitTest.java
@@ -20,12 +20,19 @@ package org.apache.hugegraph.unit;
import java.net.URI;
import java.net.URL;
import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
import org.junit.BeforeClass;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.OptionSpace;
import org.apache.hugegraph.rpc.RpcServer;
+import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.testutil.Whitebox;
+
+import com.alipay.sofa.rpc.config.ServerConfig;
+import com.alipay.sofa.rpc.server.Server;
+import io.netty.channel.ChannelFuture;
public class BaseUnitTest {
@@ -53,6 +60,34 @@ public class BaseUnitTest {
}
protected static void stopServer(RpcServer rpcServer) {
- rpcServer.destroy();
+ closeServer(rpcServer, rpcServer::destroy);
+ }
+
+ protected static void unexportServer(RpcServer rpcServer) {
+ closeServer(rpcServer, rpcServer::unexportAll);
+ }
+
+ protected static ChannelFuture listener(RpcServer rpcServer) {
+ if (!rpcServer.enabled()) {
+ return null;
+ }
+ ServerConfig config = Whitebox.getInternalState(rpcServer, "serverConfig");
+ Server server = config.getServer();
+ if (server == null || !server.isStarted()) {
+ return null;
+ }
+ Object remoting = Whitebox.getInternalState(server, "remotingServer");
+ return Whitebox.getInternalState(remoting, "channelFuture");
+ }
+
+ private static void closeServer(RpcServer rpcServer, Runnable close) {
+ ChannelFuture binding = listener(rpcServer);
+ close.run();
+ if (binding != null) {
+ // Bolt stops asynchronously; the next fixture may reuse this port.
+ ChannelFuture closed = binding.channel().closeFuture();
+ Assert.assertTrue(closed.awaitUninterruptibly(5, TimeUnit.SECONDS));
+ Assert.assertTrue(closed.isSuccess());
+ }
}
}
diff --git a/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/RpcConfigTest.java b/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/RpcConfigTest.java
index 5cdcecdfa..cf2ce94df 100644
--- a/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/RpcConfigTest.java
+++ b/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/RpcConfigTest.java
@@ -17,7 +17,9 @@
package org.apache.hugegraph.unit;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
@@ -32,15 +34,274 @@ import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.testutil.Whitebox;
import org.apache.hugegraph.unit.ServerClientTest.HelloService;
import org.apache.hugegraph.unit.ServerClientTest.HelloServiceImpl;
+import org.junit.BeforeClass;
import org.junit.Test;
+import com.alipay.sofa.rpc.bootstrap.ConsumerBootstrap;
+import com.alipay.sofa.rpc.bootstrap.DefaultConsumerBootstrap;
+import com.alipay.sofa.rpc.client.Cluster;
+import com.alipay.sofa.rpc.client.FailoverCluster;
import com.alipay.sofa.rpc.config.ConsumerConfig;
import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.ext.Extension;
+import com.alipay.sofa.rpc.ext.ExtensionLoaderFactory;
public class RpcConfigTest extends BaseUnitTest {
private static final String HELLO = HelloService.class.getName();
+ private static ConsumerFailure consumerFailure;
+
+ @BeforeClass
+ public static void registerCleanupBootstrap() {
+ ExtensionLoaderFactory.getExtensionLoader(ConsumerBootstrap.class)
+ .loadExtension(CleanupBootstrap.class);
+ ExtensionLoaderFactory.getExtensionLoader(Cluster.class)
+ .loadExtension(CleanupCluster.class);
+ }
+
+ @Test
+ public void testReferFailureRetainsCleanupOwner() {
+ this.assertReferCleanup(new IllegalStateException("refer failure"), false, false);
+ }
+
+ @Test
+ public void testReferErrorPreservedOnSelfSuppression() {
+ this.assertReferCleanup(new AssertionError("refer error"), true, false);
+ }
+
+ @Test
+ public void testCleanupErrorPromotedAboveReferException() {
+ this.assertReferCleanup(new IllegalStateException("refer failure"), false, true);
+ }
+
+ @Test
+ public void testReferFailureWithSuccessfulCleanupReleasesOwner() {
+ RpcConsumerConfig owner = new RpcConsumerConfig(new HugeConfig(Collections.emptyMap()),
+ "127.0.0.1:1");
+ ConsumerFailure control = new ConsumerFailure(owner, new IllegalStateException("refer failure"));
+ control.disarm();
+ try {
+ Assert.assertSame(control.referFailure, captureFailure(() -> owner.serviceProxy(HELLO)));
+ Assert.assertGt(0, control.destroyCalls.get());
+ Assert.assertFalse(control.retained());
+ Assert.assertNull(control.bootstrap.getProxyIns());
+ } finally {
+ control.close();
+ }
+ }
+
+ private void assertReferCleanup(Throwable primary, boolean sameCleanup, boolean fatalCleanup) {
+ RpcConsumerConfig owner = new RpcConsumerConfig(new HugeConfig(Collections.emptyMap()),
+ "127.0.0.1:1");
+ ConsumerFailure control = new ConsumerFailure(owner, primary);
+ control.unreferFailure = sameCleanup ? primary : new IllegalStateException("unrefer failure");
+ if (fatalCleanup) {
+ control.clusterFailure = new AssertionError("cluster error");
+ } else if (sameCleanup) {
+ control.clusterFailure = new AssertionError("secondary cluster error");
+ }
+ try {
+ Throwable thrown = captureFailure(() -> owner.serviceProxy(HELLO));
+ Assert.assertNotNull(control.bootstrap.getProxyIns());
+ Assert.assertSame(fatalCleanup ? control.clusterFailure : primary, thrown);
+ if (fatalCleanup) {
+ Assert.assertSame(primary, thrown.getSuppressed()[0]);
+ }
+ Assert.assertGt(0, control.destroyCalls.get());
+ Assert.assertTrue(control.retained());
+ control.disarm();
+ owner.destroy();
+ Assert.assertFalse(control.retained());
+ Assert.assertNull(control.bootstrap.getProxyIns());
+ } finally {
+ control.close();
+ }
+ }
+
+ @Test
+ public void testClusterCleanupFailureRetainsOwnerForRetry() {
+ RpcConsumerConfig owner = new RpcConsumerConfig(new HugeConfig(Collections.emptyMap()),
+ "127.0.0.1:1");
+ ConsumerFailure control = new ConsumerFailure(owner, new IllegalStateException("refer failure"));
+ control.failUnrefer = false;
+ control.clusterFailure = new IllegalStateException("cluster failure");
+ try {
+ Assert.assertSame(control.referFailure, captureFailure(() -> owner.serviceProxy(HELLO)));
+ // Native unRefer swallows a cluster Exception, so the explicit destroy must also run.
+ Assert.assertGte(2, control.destroyCalls.get());
+ Assert.assertTrue(control.retained());
+ control.disarm();
+ owner.destroy();
+ Assert.assertFalse(control.retained());
+ } finally {
+ control.close();
+ }
+ }
+
+ @Test
+ public void testDestroyContinuesRemainingConsumers() {
+ RpcConsumerConfig owner = new RpcConsumerConfig(new HugeConfig(Collections.emptyMap()),
+ "127.0.0.1:1");
+ ConsumerFailure first = new ConsumerFailure(owner, new IllegalStateException("first refer"));
+ ConsumerFailure second = null;
+ try {
+ captureFailure(() -> owner.serviceProxy(HELLO));
+ second = new ConsumerFailure(owner, new IllegalStateException("second refer"));
+ captureFailure(() -> owner.serviceProxy(HELLO));
+ second.disarm();
+ Assert.assertSame(first.unreferFailure, captureFailure(owner::removeAllServiceProxy));
+ Assert.assertNull(second.bootstrap.getProxyIns());
+ // Unrefer-only cleanup retains owners for the subsequent explicit destroy.
+ Assert.assertTrue(second.retained());
+ Assert.assertSame(first.unreferFailure, captureFailure(owner::destroy));
+ Assert.assertTrue(first.retained());
+ Assert.assertFalse(second.retained());
+ Assert.assertNull(second.bootstrap.getProxyIns());
+ first.disarm();
+ owner.destroy();
+ Assert.assertFalse(first.retained());
+ } finally {
+ first.close();
+ if (second != null) {
+ second.close();
+ }
+ }
+ }
+
+ @Test
+ public void testRemoveConsumerRetainsFailedCleanupOwner() {
+ RpcConsumerConfig owner = new RpcConsumerConfig(new HugeConfig(Collections.emptyMap()),
+ "127.0.0.1:1");
+ ConsumerFailure control = new ConsumerFailure(owner, null);
+ try {
+ HelloService proxy = owner.serviceProxy(HELLO);
+ Assert.assertSame(control.unreferFailure, captureFailure(() -> {
+ owner.removeServiceProxy(null, HELLO, proxy);
+ }));
+ Assert.assertGt(0, control.destroyCalls.get());
+ Assert.assertTrue(control.retained());
+ Map<String, ConsumerConfig<?>> configs = Whitebox.getInternalState(owner, "configs");
+ Assert.assertSame(control.bootstrap.getConsumerConfig(), configs.get(HELLO));
+ control.disarm();
+ owner.destroy();
+ Assert.assertFalse(control.retained());
+ } finally {
+ control.close();
+ }
+ }
+
+ private static Throwable captureFailure(Runnable action) {
+ try {
+ action.run();
+ } catch (RuntimeException | Error failure) {
+ return failure;
+ }
+ throw new AssertionError("Expected consumer failure");
+ }
+
+ private static final class ConsumerFailure implements AutoCloseable {
+
+ private static final AtomicInteger IDS = new AtomicInteger();
+ private final RpcConsumerConfig owner;
+ private final Throwable referFailure;
+ private final AtomicInteger destroyCalls = new AtomicInteger();
+ private Throwable unreferFailure = new IllegalStateException("unrefer failure");
+ private Throwable clusterFailure;
+ private boolean failUnrefer = true;
+ private CleanupBootstrap bootstrap;
+
+ private ConsumerFailure(RpcConsumerConfig owner, Throwable referFailure) {
+ this.owner = owner;
+ this.referFailure = referFailure;
+ ConsumerConfig<HelloService> config = new ConsumerConfig<>();
+ config.setInterfaceId(HELLO).setUniqueId("cleanup-" + IDS.incrementAndGet())
+ .setProtocol("bolt").setDirectUrl("127.0.0.1:1")
+ .setLazy(true).setCheck(false).setInJVM(false)
+ .setBootstrap("hugegraph_cleanup_test").setCluster("hugegraph_cleanup_test");
+ Map<String, ConsumerConfig<?>> configs = Whitebox.getInternalState(owner, "configs");
+ configs.put(HELLO, config);
+ consumerFailure = this;
+ }
+
+ private boolean retained() {
+ List<?> bootstraps = Whitebox.getInternalState(this.owner, "bootstraps");
+ return bootstraps.contains(this.bootstrap);
+ }
+
+ private void disarm() {
+ this.failUnrefer = false;
+ this.clusterFailure = null;
+ }
+
+ @Override
+ public void close() {
+ this.disarm();
+ if (this.bootstrap != null) {
+ this.bootstrap.unRefer();
+ this.bootstrap.getCluster().destroy();
+ }
+ consumerFailure = null;
+ }
+ }
+
+ @Extension("hugegraph_cleanup_test")
+ public static final class CleanupBootstrap extends DefaultConsumerBootstrap<HelloService> {
+
+ private final ConsumerFailure control;
+
+ public CleanupBootstrap(ConsumerConfig<HelloService> config) {
+ super(config);
+ this.control = consumerFailure;
+ this.control.bootstrap = this;
+ }
+
+ @Override
+ public HelloService refer() {
+ HelloService proxy = super.refer();
+ if (this.control.referFailure != null) {
+ throwFailure(this.control.referFailure);
+ }
+ return proxy;
+ }
+
+ @Override
+ public void unRefer() {
+ if (this.control.failUnrefer) {
+ throwFailure(this.control.unreferFailure);
+ }
+ super.unRefer();
+ }
+
+ private static void throwFailure(Throwable failure) {
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw (RuntimeException) failure;
+ }
+ }
+
+ @Extension("hugegraph_cleanup_test")
+ public static final class CleanupCluster extends FailoverCluster {
+
+ private final ConsumerFailure control;
+
+ public CleanupCluster(ConsumerBootstrap<?> bootstrap) {
+ super(bootstrap);
+ this.control = ((CleanupBootstrap) bootstrap).control;
+ }
+
+ @Override
+ public void destroy() {
+ // Use the native cluster lifecycle; control only the cleanup failure boundary.
+ this.control.destroyCalls.incrementAndGet();
+ if (this.control.clusterFailure != null) {
+ CleanupBootstrap.throwFailure(this.control.clusterFailure);
+ }
+ super.destroy();
+ }
+ }
+
@Test
public void testExcludeSelfUrl() {
Assert.assertEquals("127.0.0.1:8092,127.0.0.1:8093",
@@ -171,13 +432,41 @@ public class RpcConfigTest extends BaseUnitTest {
Assert.assertEquals("g1", graph.getUniqueId());
Assert.assertEquals("fanout", graph.getCluster());
- // Configs are cached per service id
- Assert.assertSame(plain, consumerConfig(consumer, null, HELLO));
- Assert.assertSame(graph, consumerConfig(consumer, "g1", HELLO));
- Assert.assertNotSame(graph, consumerConfig(consumer, "g2", HELLO));
consumer.destroy();
}
+ @Test
+ public void testConsumerConfigCachedThroughServiceProxy() {
+ RpcServer server = new RpcServer(config("server-random"));
+ RpcConsumerConfig consumer = null;
+ try {
+ server.config().addService(HelloService.class, new HelloServiceImpl());
+ server.config().addService("g1", HelloService.class, new HelloServiceImpl());
+ server.config().addService("g2", HelloService.class, new HelloServiceImpl());
+ server.exportAll();
+ consumer = new RpcConsumerConfig(config("client"), server.host() + ":" + server.port());
+ ConsumerConfig<?> plain = referredConfig(consumer, null);
+ ConsumerConfig<?> graph = referredConfig(consumer, "g1");
+ Assert.assertSame(plain, referredConfig(consumer, null));
+ Assert.assertSame(graph, referredConfig(consumer, "g1"));
+ Assert.assertNotSame(graph, referredConfig(consumer, "g2"));
+ } finally {
+ if (consumer != null) {
+ consumer.destroy();
+ }
+ stopServer(server);
+ }
+ }
+
+ private static ConsumerConfig<?> referredConfig(RpcConsumerConfig consumer, String graph) {
+ HelloService proxy = consumer.serviceProxy(graph, HELLO);
+ Assert.assertEquals("hello tom!", proxy.hello("tom"));
+ Map<String, ConsumerConfig<?>> configs = Whitebox.getInternalState(consumer, "configs");
+ String id = graph == null ? HELLO : HELLO + ":" + graph;
+ Assert.assertNotNull(configs.get(id));
+ return configs.get(id);
+ }
+
@Test
public void testServerDestroyClosesServices() {
RpcServer server = new RpcServer(config("server-random"));
diff --git a/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/ServerClientTest.java b/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/ServerClientTest.java
index 07eb893c7..cf138721d 100644
--- a/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/ServerClientTest.java
+++ b/hugegraph-commons/hugegraph-rpc/src/test/java/org/apache/hugegraph/unit/ServerClientTest.java
@@ -22,6 +22,13 @@ import java.net.InetAddress;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.rpc.RpcClientProvider;
@@ -38,9 +45,12 @@ import org.junit.BeforeClass;
import org.junit.Test;
import com.alipay.sofa.rpc.common.RpcOptions;
+import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.config.ServerConfig;
import com.alipay.sofa.rpc.core.exception.SofaRpcException;
import com.alipay.sofa.rpc.core.exception.SofaRpcRuntimeException;
import com.google.common.collect.ImmutableMap;
+import io.netty.channel.ChannelFuture;
public class ServerClientTest extends BaseUnitTest {
@@ -59,7 +69,7 @@ public class ServerClientTest extends BaseUnitTest {
rpcClient.destroy();
}
if (rpcServer != null) {
- rpcServer.destroy();
+ stopServer(rpcServer);
}
}
@@ -69,7 +79,45 @@ public class ServerClientTest extends BaseUnitTest {
rpcClient.unreferAll();
}
if (rpcServer != null) {
- rpcServer.unexportAll();
+ unexportServer(rpcServer);
+ }
+ }
+
+ @Test
+ public void testUnexportWaitsBeforeReusingListenerPort() throws Exception {
+ rpcServer.config().addService(HelloService.class, new HelloServiceImpl());
+ startServer(rpcServer);
+ ChannelFuture binding = listener(rpcServer);
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ binding.channel().eventLoop().execute(() -> {
+ entered.countDown();
+ try {
+ E.checkState(release.await(5, TimeUnit.SECONDS), "RPC close barrier timed out");
+ } catch (InterruptedException failure) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(failure);
+ }
+ });
+ try {
+ Assert.assertTrue(entered.await(5, TimeUnit.SECONDS));
+ Future<?> closing = worker.submit(() -> unexportServer(rpcServer));
+ Assert.assertThrows(TimeoutException.class, () -> closing.get(1, TimeUnit.SECONDS));
+ Assert.assertTrue(rpcServer.config().configs().isEmpty());
+ Assert.assertTrue(binding.channel().isOpen());
+ Assert.assertFalse(binding.channel().closeFuture().isDone());
+ release.countDown();
+ closing.get(5, TimeUnit.SECONDS);
+ Assert.assertTrue(binding.channel().closeFuture().isSuccess());
+ rpcServer.config().addService(HelloService.class, new HelloServiceImpl());
+ startServer(rpcServer);
+ HelloService client = rpcClient.config().serviceProxy(HelloService.class);
+ Assert.assertEquals("hello tom!", client.hello("tom"));
+ } finally {
+ release.countDown();
+ worker.shutdownNow();
+ Assert.assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
}
}
@@ -603,26 +651,64 @@ public class ServerClientTest extends BaseUnitTest {
@Test
public void testExportMultiTimesOfSameServer() {
RpcServer rpcServerExport = new RpcServer(config("server-random"));
- rpcServerExport.config().addService(HelloService.class,
- new HelloServiceImpl());
- rpcServerExport.exportAll();
- rpcServerExport.exportAll();
-
- stopServer(rpcServerExport);
+ try {
+ rpcServerExport.config().addService(HelloService.class, new HelloServiceImpl());
+ rpcServerExport.exportAll();
+ rpcServerExport.exportAll();
+ Assert.assertEquals(1, rpcServerExport.config().configs().get(
+ HelloService.class.getName()).getServer().size());
+ } finally {
+ stopServer(rpcServerExport);
+ }
}
@Test
public void testExportMultiTimesOfSameService() {
RpcServer rpcServerExport = new RpcServer(config("server-random"));
- rpcServerExport.config().addService(HelloService.class,
- new HelloServiceImpl());
- rpcServerExport.exportAll();
-
- rpcServerExport.config().addService("graph", HelloService.class,
- new HelloServiceImpl());
- rpcServerExport.exportAll();
+ try {
+ rpcServerExport.config().addService(HelloService.class, new HelloServiceImpl());
+ rpcServerExport.exportAll();
+ rpcServerExport.config().addService("graph", HelloService.class, new HelloServiceImpl());
+ rpcServerExport.exportAll();
+ rpcServerExport.exportAll();
+ for (ProviderConfig<?> provider : rpcServerExport.config().configs().values()) {
+ Assert.assertEquals(1, provider.getServer().size());
+ }
+ } finally {
+ stopServer(rpcServerExport);
+ }
+ }
- stopServer(rpcServerExport);
+ @Test
+ public void testExportRetriesWithoutDuplicatingConfiguredServers() {
+ RpcServer server = new RpcServer(config("server-random"));
+ AtomicInteger exports = new AtomicInteger();
+ IllegalStateException failure = new IllegalStateException("export failed before binding");
+ ProviderConfig<HelloService> provider = new ProviderConfig<>() {
+ @Override
+ public void export() {
+ if (exports.incrementAndGet() == 1) {
+ throw failure;
+ }
+ }
+ };
+ // Sofa equality ignores settings such as serialization; preserve distinct config objects.
+ ServerConfig existing = new ServerConfig().setProtocol("bolt").setHost("127.0.0.1")
+ .setPort(0).setSerialization("java");
+ provider.setServer(existing);
+ server.config().configs().put(HelloService.class.getName(), provider);
+ try {
+ Assert.assertThrows(IllegalStateException.class, server::exportAll,
+ caught -> Assert.assertSame(failure, caught));
+ server.exportAll();
+ server.exportAll();
+ Assert.assertEquals(3, exports.get());
+ Assert.assertEquals(2, provider.getServer().size());
+ Assert.assertSame(existing, provider.getServer().get(0));
+ Assert.assertSame(Whitebox.getInternalState(server, "serverConfig"), provider.getServer().get(1));
+ } finally {
+ stopServer(server);
+ }
}
@Test
@@ -711,6 +797,108 @@ public class ServerClientTest extends BaseUnitTest {
});
}
+ @Test
+ public void testRemoveServiceUsesImplementationIdentity() throws Exception {
+ RpcServer server = new RpcServer(config("server-random"));
+ GraphHelloServiceImpl original = new GraphHelloServiceImpl("graph");
+ RpcProviderConfig providers = server.config();
+ String serviceId = providers.addService("graph", HelloService.class, original);
+ PausedProvider provider = new PausedProvider();
+ provider.setId(serviceId).setUniqueId("graph");
+ provider.setInterfaceId(HelloService.class.getName()).setRef(original);
+ providers.configs().put(serviceId, provider);
+ server.exportAll();
+ RpcClientProvider client = clientFor(server);
+ ExecutorService worker = Executors.newFixedThreadPool(2);
+ try {
+ HelloService proxy = client.config().serviceProxy("graph", HelloService.class);
+ Assert.assertFalse(providers.removeService(serviceId, new GraphHelloServiceImpl("graph")));
+ Assert.assertEquals("graph: hello tom!", proxy.hello("tom"));
+ provider.armed = true;
+ Future<?> exporting = worker.submit(server::exportAll);
+ Assert.assertTrue(provider.entered.await(5, TimeUnit.SECONDS));
+ CountDownLatch removing = new CountDownLatch(1);
+ Future<Boolean> removal = worker.submit(() -> {
+ removing.countDown();
+ return providers.removeService(serviceId, original);
+ });
+ Assert.assertTrue(removing.await(5, TimeUnit.SECONDS));
+ Assert.assertThrows(TimeoutException.class, () -> removal.get(1, TimeUnit.SECONDS));
+ Assert.assertSame(provider, providers.configs().get(serviceId));
+ provider.release.countDown();
+ exporting.get(5, TimeUnit.SECONDS);
+ Assert.assertTrue(removal.get(5, TimeUnit.SECONDS));
+ Assert.assertFalse(providers.configs().containsKey(serviceId));
+ GraphHelloServiceImpl replacement = new GraphHelloServiceImpl("replacement");
+ providers.addService("graph", HelloService.class, replacement);
+ server.exportAll();
+ Assert.assertFalse(providers.removeService(serviceId, original));
+ Assert.assertSame(replacement, providers.configs().get(serviceId).getRef());
+ Assert.assertEquals("replacement: hello tom!", proxy.hello("tom"));
+ } finally {
+ provider.release.countDown();
+ worker.shutdownNow();
+ try {
+ Assert.assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ } finally {
+ client.destroy();
+ stopServer(server);
+ }
+ }
+ }
+
+ private static class PausedProvider extends ProviderConfig<HelloService> {
+
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private volatile boolean armed;
+
+ @Override
+ public ProviderConfig<HelloService> setTimeout(int timeout) {
+ if (this.armed) {
+ this.entered.countDown();
+ try {
+ E.checkState(this.release.await(5, TimeUnit.SECONDS), "RPC export barrier timed out");
+ } catch (InterruptedException failure) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(failure);
+ }
+ this.armed = false;
+ }
+ return super.setTimeout(timeout);
+ }
+ }
+
+ @Test
+ public void testRemoveServiceProxyUsesProxyIdentity() {
+ RpcServer server = new RpcServer(config("server-random"));
+ server.config().addService("graph", HelloService.class, new GraphHelloServiceImpl("graph"));
+ server.config().addService("other", HelloService.class, new GraphHelloServiceImpl("other"));
+ server.exportAll();
+ RpcClientProvider client = clientFor(server);
+ try {
+ RpcConsumerConfig consumers = client.config();
+ HelloService original = consumers.serviceProxy("graph", HelloService.class);
+ HelloService second = consumers.serviceProxy("graph", HelloService.class);
+ HelloService other = consumers.serviceProxy("other", HelloService.class);
+ Assert.assertNotSame(original, second);
+ Assert.assertTrue(consumers.removeServiceProxy("graph", HelloService.class.getName(), original));
+ Assert.assertThrows(SofaRpcRuntimeException.class, () -> original.hello("tom"));
+ Assert.assertEquals("graph: hello tom!", second.hello("tom"));
+ Assert.assertFalse(consumers.removeServiceProxy("graph", HelloService.class.getName(), original));
+ Assert.assertTrue(consumers.removeServiceProxy("graph", HelloService.class.getName(), second));
+ List<?> bootstraps = Whitebox.getInternalState(consumers, "bootstraps");
+ Assert.assertEquals(1, bootstraps.size());
+ HelloService replacement = consumers.serviceProxy("graph", HelloService.class);
+ Assert.assertFalse(consumers.removeServiceProxy("graph", HelloService.class.getName(), original));
+ Assert.assertEquals("graph: hello tom!", replacement.hello("tom"));
+ Assert.assertEquals("other: hello tom!", other.hello("tom"));
+ } finally {
+ client.destroy();
+ stopServer(server);
+ }
+ }
+
@Test
public void testServerDisabled() {
HugeConfig clientConf = config(false);
diff --git a/hugegraph-server/README.md b/hugegraph-server/README.md
index ee5f8cff6..f559004bb 100644
--- a/hugegraph-server/README.md
+++ b/hugegraph-server/README.md
@@ -16,6 +16,8 @@ HugeGraph Server consists of two layers of functionality: the graph engine layer
See [launcher GC options](../docs/server-runtime.md#launcher-gc-options) for explicit collector selection
and caller JVM tuning.
+See [cache RPC graph lifecycle](../docs/server-runtime.md#cache-rpc-graph-lifecycle) for dynamic
+graph registration, resource cleanup and notification delivery limits.
See [local loading ownership](../docs/server-runtime.md#local-loading-ownership) for
concurrent import mode protection and its compatibility limits.
See [pairwise same neighbors](../docs/same-neighbors-batch.md) for the batch traverser request and response.
diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
index 1967ed072..7bd04a2be 100644
--- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
+++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
@@ -919,6 +919,12 @@ public final class HugeGraphAuthProxy implements HugeGraph {
this.hugegraph.registerRpcServices(serverConfig, clientConfig);
}
+ @Override
+ public void unregisterRpcServices() {
+ this.verifyAdminPermission();
+ this.hugegraph.unregisterRpcServices();
+ }
+
@Override
public void initBackend() {
this.verifyAdminPermission();
diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java
index f24bf6ff1..2a8c60816 100644
--- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java
+++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java
@@ -30,6 +30,7 @@ import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -136,6 +137,8 @@ import org.apache.tinkerpop.gremlin.structure.Transaction;
import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
import org.slf4j.Logger;
+import com.alipay.sofa.jraft.NodeManager;
+import com.alipay.sofa.jraft.util.Endpoint;
import com.alipay.sofa.rpc.config.ServerConfig;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
@@ -170,6 +173,11 @@ public final class GraphManager {
private final Map<String, Graph> graphs;
private final Set<String> localGraphs;
private final Set<String> removingGraphs;
+ // Guarded by this manager monitor; placement-event tokens are consumed independently.
+ private final Set<HugeGraph> rpcRemovingOwners;
+ private final Set<String> pendingGraphAdds;
+ private final Set<HugeGraph> failedPeerRemovals;
+ private final Map<String, HugeGraph> clearedGraphs;
private final Set<String> creatingGraphs;
private final HugeAuthenticator authenticator;
private final AuthManager authManager;
@@ -181,6 +189,8 @@ public final class GraphManager {
private final RpcServer rpcServer;
private final RpcClientProvider rpcClient;
+ private boolean rpcInitialized;
+ private volatile Endpoint raftRpcEndpoint;
private final GlobalMasterInfo globalNodeRoleInfo;
private final HugeConfig conf;
private final EventHub eventHub;
@@ -216,6 +226,10 @@ public final class GraphManager {
// key is graphSpaceName + "-" + graphName
this.graphs = new ConcurrentHashMap<>();
this.removingGraphs = ConcurrentHashMap.newKeySet();
+ this.rpcRemovingOwners = Collections.newSetFromMap(new IdentityHashMap<>());
+ this.pendingGraphAdds = new HashSet<>();
+ this.failedPeerRemovals = Collections.newSetFromMap(new IdentityHashMap<>());
+ this.clearedGraphs = new ConcurrentHashMap<>();
this.creatingGraphs = ConcurrentHashMap.newKeySet();
this.authenticator = HugeAuthenticator.loadAuthenticator(conf);
this.serviceGraphSpace = conf.get(ServerOptions.SERVICE_GRAPH_SPACE);
@@ -702,7 +716,7 @@ public final class GraphManager {
// Remove graphs from GraphManager
for (String graph : this.graphs.keySet()) {
String[] parts = graph.split(DELIMITER);
- this.dropGraph(parts[0], parts[1], false);
+ this.dropGraph(parts[0], parts[1], false, false);
}
int count = 0;
while (!this.graphs.isEmpty() && count++ < 10) {
@@ -1451,19 +1465,35 @@ public final class GraphManager {
// Init graph and start it
graph.create(this.graphsDir, this.globalNodeRoleInfo);
- // Let gremlin server and rest server add graph to context
+ // Complete RPC registration before Gremlin listeners expose the graph.
+ this.publishGraph(graph);
this.notifyEvent(Events.GRAPH_CREATE, graph);
} catch (Throwable e) {
LOG.error("Failed to create graph '{}' due to: {}",
name, e.getMessage(), e);
if (graph != null) {
+ synchronized (this) {
+ if (this.graphs.get(graph.spaceGraphName()) == graph) {
+ this.rpcRemovingOwners.add(graph);
+ }
+ }
try {
- this.dropGraphLocal(graph);
- this.graphs.remove(graph.spaceGraphName(), graph);
+ this.unregisterRpcServices(graph);
+ HugeGraph candidate = graph;
+ HugeGraphAuthProxy.runAsAdmin(() -> this.dropGraphLocal(candidate));
+ synchronized (this) {
+ this.graphs.remove(graph.spaceGraphName(), graph);
+ this.rpcRemovingOwners.remove(graph);
+ }
this.notifyEventLenient(Events.GRAPH_DROP, graph);
} catch (Throwable cleanup) {
e.addSuppressed(cleanup);
- this.graphs.putIfAbsent(graph.spaceGraphName(), graph);
+ synchronized (this) {
+ this.graphs.putIfAbsent(graph.spaceGraphName(), graph);
+ if (this.graphs.get(graph.spaceGraphName()) == graph) {
+ this.rpcRemovingOwners.add(graph);
+ }
+ }
}
}
throw e;
@@ -1597,7 +1627,6 @@ public final class GraphManager {
graph.updateTime(timeStamp);
String graphName = spaceGraphName(graphSpace, name);
- this.graphs.put(graphName, graph);
/*
* Let gremlin server and rest server context add graph before the
@@ -1605,16 +1634,10 @@ public final class GraphManager {
* graph behind in meta for the other servers to converge on
*/
try {
+ this.publishGraph(graph);
this.notifyEvent(Events.GRAPH_CREATE, graph);
} catch (Throwable e) {
- try {
- graph.close();
- this.notifyEventLenient(Events.GRAPH_DROP, graph);
- this.graphs.remove(graphName, graph);
- HugeFactory.remove(graph);
- } catch (Exception e1) {
- e.addSuppressed(e1);
- }
+ this.rollbackGraph(graphName, graph, e);
throw e;
}
@@ -1712,9 +1735,15 @@ public final class GraphManager {
}
public void close() {
- for (Graph graph : this.graphs.values()) {
+ for (Map.Entry<String, Graph> entry : this.graphs.entrySet()) {
+ Graph graph = entry.getValue();
try {
graph.close();
+ synchronized (this) {
+ if (this.clearedGraphs.get(entry.getKey()) == graph) {
+ this.clearedGraphs.remove(entry.getKey());
+ }
+ }
} catch (Throwable e) {
LOG.warn("Failed to close graph '{}'", graph, e);
}
@@ -1723,7 +1752,7 @@ public final class GraphManager {
this.unlistenChanges();
}
- private void startRpcServer() {
+ private synchronized void startRpcServer() {
if (!this.rpcServer.enabled()) {
LOG.info("RpcServer is not enabled, skip starting rpc service");
return;
@@ -1737,20 +1766,50 @@ public final class GraphManager {
this.authenticator.authManager());
}
- // Start graph rpc service if RPC_REMOTE_URL enabled
- if (this.rpcClient.enabled()) {
- RpcConsumerConfig clientConfig = this.rpcClient.config();
-
- for (Graph graph : this.graphs.values()) {
- HugeGraph hugegraph = (HugeGraph) graph;
- hugegraph.registerRpcServices(serverConfig, clientConfig);
+ try {
+ // Start graph rpc service if RPC_REMOTE_URL enabled
+ if (this.rpcClient.enabled()) {
+ RpcConsumerConfig clientConfig = this.rpcClient.config();
+
+ for (Graph graph : this.graphs.values()) {
+ HugeGraph hugegraph = (HugeGraph) graph;
+ HugeGraphAuthProxy.runAsAdmin(() -> {
+ // Auth wrappers and the owner must use the same existing lifecycle monitor.
+ synchronized (hugegraph.hugegraph()) {
+ if (hugegraph.closed() || this.rpcRemovingOwners.contains(hugegraph)) {
+ // Retained provider configs must be removed before exportAll.
+ this.unregisterRpcServices(hugegraph);
+ return;
+ }
+ hugegraph.registerRpcServices(serverConfig, clientConfig);
+ }
+ });
+ }
}
- }
- try {
this.rpcServer.exportAll();
+ this.rpcInitialized = true;
} catch (Throwable e) {
- this.rpcServer.destroy();
+ this.rpcInitialized = false;
+ for (Graph graph : this.graphs.values()) {
+ if (this.rpcRemovingOwners.contains(graph)) {
+ continue;
+ }
+ try {
+ this.unregisterRpcServices((HugeGraph) graph);
+ } catch (Throwable cleanup) {
+ if (e != cleanup) {
+ e.addSuppressed(cleanup);
+ }
+ }
+ }
+ try {
+ this.rpcServer.destroy();
+ } catch (Throwable cleanup) {
+ if (e != cleanup) {
+ e.addSuppressed(cleanup);
+ }
+ }
throw e;
}
}
@@ -1765,8 +1824,10 @@ public final class GraphManager {
if (!serverConfig.getServer().isStarted()) {
serverConfig.getServer().start();
}
- return Whitebox.getInternalState(serverConfig.getServer(),
- "remotingServer");
+ com.alipay.remoting.rpc.RpcServer server = Whitebox.getInternalState(serverConfig.getServer(),
+ "remotingServer");
+ this.raftRpcEndpoint = new Endpoint(server.ip(), server.port());
+ return server;
}
private void destroyRpcServer() {
@@ -1774,6 +1835,11 @@ public final class GraphManager {
this.rpcClient.destroy();
} finally {
this.rpcServer.destroy();
+ Endpoint endpoint = this.raftRpcEndpoint;
+ if (endpoint != null) {
+ NodeManager.getInstance().removeAddress(endpoint);
+ this.raftRpcEndpoint = null;
+ }
}
}
@@ -1821,7 +1887,12 @@ public final class GraphManager {
this.transferPdPeersConfig(config);
Graph graph = GraphFactory.open(config);
- this.graphs.put(defaultSpaceGraphName(name), graph);
+ try {
+ this.publishGraph(defaultSpaceGraphName(name), (HugeGraph) graph);
+ } catch (RuntimeException | Error failure) {
+ this.rollbackGraph(defaultSpaceGraphName(name), (HugeGraph) graph, failure);
+ throw failure;
+ }
HugeConfig graphConfig = (HugeConfig) graph.configuration();
// A graph Gremlin Server opened first from this path is cached without a file
@@ -1840,6 +1911,40 @@ public final class GraphManager {
}
}
+ private void rollbackGraph(String graphName, HugeGraph graph, Throwable failure) {
+ synchronized (this) {
+ if (this.graphs.get(graphName) == graph) {
+ this.rpcRemovingOwners.add(graph);
+ }
+ }
+ try {
+ HugeGraphAuthProxy.runAsAdmin(() -> {
+ try {
+ graph.close();
+ } catch (Exception cleanup) {
+ throw new HugeException("Failed to close unpublished graph '%s'", cleanup, graphName);
+ }
+ });
+ this.notifyEventLenient(Events.GRAPH_DROP, graph);
+ synchronized (this) {
+ HugeFactory.remove(graph);
+ this.graphs.remove(graphName, graph);
+ this.rpcRemovingOwners.remove(graph);
+ }
+ } catch (Throwable cleanup) {
+ if (cleanup != failure) {
+ failure.addSuppressed(cleanup);
+ }
+ // Pending owners must keep their graph/factory identity for cleanup retry.
+ synchronized (this) {
+ this.graphs.putIfAbsent(graphName, graph);
+ if (this.graphs.get(graphName) == graph) {
+ this.rpcRemovingOwners.add(graph);
+ }
+ }
+ }
+ }
+
/*
* Gremlin Server opens the graphs in conf/graphs once, while it is being
* prepared, and drops any graph whose open fails there (for example when
@@ -1987,18 +2092,65 @@ public final class GraphManager {
LOG.debug("RestServer accepts event '{}'", event.name());
event.checkArgs(HugeGraph.class);
HugeGraph graph = (HugeGraph) event.args()[0];
- this.graphs.put(graph.spaceGraphName(), graph);
+ this.publishGraph(graph);
return null;
});
this.eventHub.listen(Events.GRAPH_DROP, event -> {
LOG.debug("RestServer accepts event '{}'", event.name());
event.checkArgs(HugeGraph.class);
HugeGraph graph = (HugeGraph) event.args()[0];
- this.graphs.remove(graph.spaceGraphName(), graph);
+ this.unpublishGraph(graph);
return null;
});
}
+ private void publishGraph(HugeGraph graph) {
+ this.publishGraph(graph.spaceGraphName(), graph);
+ }
+
+ private synchronized void publishGraph(String name, HugeGraph graph) {
+ if (this.graphs.get(name) == graph) {
+ return;
+ }
+ this.registerRpcServices(graph);
+ this.graphs.put(name, graph);
+ }
+
+ private void registerRpcServices(HugeGraph graph) {
+ if (this.rpcInitialized && this.rpcServer.enabled() && this.rpcClient.enabled()) {
+ try {
+ HugeGraphAuthProxy.runAsAdmin(() -> graph.registerRpcServices(this.rpcServer.config(),
+ this.rpcClient.config()));
+ this.rpcServer.exportAll();
+ } catch (RuntimeException | Error failure) {
+ try {
+ this.unregisterRpcServices(graph);
+ } catch (RuntimeException | Error cleanup) {
+ failure.addSuppressed(cleanup);
+ }
+ throw failure;
+ }
+ }
+ }
+
+ private void unregisterRpcServices(HugeGraph graph) {
+ HugeGraphAuthProxy.runAsAdmin(graph::unregisterRpcServices);
+ }
+
+ private synchronized boolean unpublishGraph(HugeGraph graph) {
+ String name = graph.spaceGraphName();
+ if (this.graphs.get(name) != graph) {
+ return false;
+ }
+ // A failed cleanup keeps the owner published so that drop can retry.
+ this.unregisterRpcServices(graph);
+ boolean removed = this.graphs.remove(name, graph);
+ if (removed) {
+ this.rpcRemovingOwners.remove(graph);
+ }
+ return removed;
+ }
+
private void unlistenChanges() {
this.eventHub.unlisten(Events.GRAPH_CREATE);
this.eventHub.unlisten(Events.GRAPH_DROP);
@@ -2236,7 +2388,6 @@ public final class GraphManager {
creator, config, false);
graph1.createTime(createTime);
graph1.updateTime(updateTime);
- this.graphs.put(key, graph1);
return graph1;
}
throw new NotFoundException(String.format("Graph '%s' does not exist", name));
@@ -2257,21 +2408,44 @@ public final class GraphManager {
"The graph '%s' is the only one, not allowed to delete",
name);
- this.dropGraphLocal(graph);
+ try {
+ this.dropGraphLocal(graph);
+ } catch (RuntimeException | Error failure) {
+ synchronized (this) {
+ if (graph.closed()) {
+ this.rpcRemovingOwners.add(graph);
+ }
+ }
+ // Local clear failure leaves the still-open owner's existing RPC services untouched.
+ throw failure;
+ }
// Let gremlin server and rest server context remove graph
this.notifyEventLenient(Events.GRAPH_DROP, graph);
}
public void dropGraph(String graphSpace, String name, boolean clear) {
+ this.dropGraph(graphSpace, name, clear, true);
+ }
+
+ private void dropGraph(String graphSpace, String name, boolean clear, boolean replayPendingAdd) {
if (!isPDEnabled()) {
dropGraphLocal(name);
return;
}
+ HugeGraph graph = this.graph(graphSpace, name);
+ E.checkArgumentNotNull(graph, "The graph '%s' doesn't exist", name);
+ this.dropGraph(graphSpace, name, clear, graph, replayPendingAdd);
+ }
+
+ private void dropGraph(String graphSpace, String name, boolean clear, HugeGraph g, boolean replayPendingAdd) {
+ synchronized (this) {
+ if (g == null || this.graphs.get(spaceGraphName(graphSpace, name)) != g) {
+ return;
+ }
+ }
boolean grpcThread = Thread.currentThread().getName().contains("grpc");
- HugeGraph g = this.graph(graphSpace, name);
- E.checkArgumentNotNull(g, "The graph '%s' doesn't exist", name);
if (this.localGraphs.contains(name)) {
throw new HugeException("Can't delete graph '%s' loaded from " +
"local config. Please delete config file " +
@@ -2280,7 +2454,13 @@ public final class GraphManager {
}
String graphName = spaceGraphName(graphSpace, name);
- if (clear) {
+ if (clear && g.closed() && this.clearedGraphs.get(graphName) != g) {
+ // A non-clearing close cannot certify that destructive deletion completed.
+ throw new HugeException("Graph '%s' is closed without a successful backend clear; " +
+ "retain its owner and complete backend cleanup before deletion", graphName);
+ }
+ // The same owner may await close after clear succeeded, even while a foreign tx remains open.
+ if (clear && !g.closed() && this.clearedGraphs.get(graphName) != g) {
// The auth proxy checks ADMIN before the local deletion fence.
TaskScheduler scheduler = g.taskScheduler();
if (!scheduler.close()) {
@@ -2296,54 +2476,98 @@ public final class GraphManager {
"Retry deletion after owners exit, or restart to resume task admission",
graphName);
}
- this.removingGraphs.add(graphName);
+ // Cleanup failure must precede metadata removal and destructive clear.
+ synchronized (this) {
+ this.rpcRemovingOwners.add(g);
+ this.unregisterRpcServices(g);
+ this.removingGraphs.add(graphName);
+ }
+ boolean metadataRemoved = false;
try {
this.metaManager.removeGraphConfig(graphSpace, name);
+ metadataRemoved = true;
this.metaManager.notifyGraphRemove(graphSpace, name);
} catch (Exception e) {
+ // No backend clear has started; retain RPC for the same open local owner.
+ synchronized (this) {
+ if (!metadataRemoved) {
+ // No local REMOVE was sent; a later peer REMOVE must still be handled.
+ this.removingGraphs.remove(graphName);
+ }
+ if (this.graphs.get(graphName) == g && !g.closed()) {
+ try {
+ this.registerRpcServices(g);
+ this.rpcRemovingOwners.remove(g);
+ } catch (RuntimeException | Error restoration) {
+ if (e != restoration) {
+ e.addSuppressed(restoration);
+ }
+ }
+ }
+ }
throw new HugeException(
- "Failed to remove graph config of '%s'", name, e);
+ "Failed to remove graph config of '%s'", e, name);
}
g.clearBackend();
- try {
- g.close();
- } catch (Exception e) {
- if (!g.closed()) {
- throw new HugeException("Failed to close graph '%s', please retry later", e, graphName);
+ synchronized (this) {
+ if (this.graphs.get(graphName) == g) {
+ this.clearedGraphs.put(graphName, g);
}
- LOG.warn("Failed to close graph", e);
}
}
- GraphSpace gs = this.graphSpace(graphSpace);
- if (!grpcThread) {
- gs.recycleGraph();
- LOG.info("The graph_number_used successfully decreased to {} " +
- "of graph space: {} for graph: {}",
- gs.graphNumberUsed(), gs.name(), name);
- this.metaManager.updateGraphSpaceConfig(graphSpace, gs);
+ try {
+ g.close();
+ } catch (Exception failure) {
+ // Logical closed does not certify native release; keep this owner until close succeeds.
+ throw new HugeException("Failed to close graph '%s', please retry later", failure, graphName);
+ }
+ // Serialize with publication, but leave graph close outside the manager gate.
+ boolean replayAdd = false;
+ synchronized (this) {
+ if (this.graphs.get(graphName) == g) {
+ this.unregisterRpcServices(g);
+ GraphSpace gs = this.graphSpace(graphSpace);
+ if (!grpcThread) {
+ int used = gs.graphNumberUsed();
+ gs.recycleGraph();
+ LOG.info("The graph_number_used successfully decreased to {} " +
+ "of graph space: {} for graph: {}",
+ gs.graphNumberUsed(), gs.name(), name);
+ try {
+ this.metaManager.updateGraphSpaceConfig(graphSpace, gs);
+ } catch (RuntimeException | Error failure) {
+ gs.graphNumberUsed(used);
+ throw failure;
+ }
+ }
+ // Remove the factory binding before permitting same-name creation.
+ HugeFactory.remove(g);
+ this.graphs.remove(graphName, g);
+ this.rpcRemovingOwners.remove(g);
+ this.failedPeerRemovals.remove(g);
+ if (replayPendingAdd) {
+ replayAdd = this.pendingGraphAdds.contains(graphName);
+ } else {
+ // Reload loads all current metadata after every old owner has been removed.
+ this.pendingGraphAdds.remove(graphName);
+ }
+ if (this.clearedGraphs.get(graphName) == g) {
+ this.clearedGraphs.remove(graphName);
+ }
+ }
}
// Let gremlin server and rest server context remove graph
LOG.info("Notify remove graph {} by GRAPH_DROP event", name);
- Graph graph = this.graphs.get(graphName);
- if (graph != null) {
- try {
- graph.close();
- } catch (Exception e) {
- if (!((HugeGraph) graph).closed()) {
- throw new HugeException("Failed to close graph '%s', please retry later", e, graphName);
- }
- LOG.warn("Failed to close graph", e);
- }
- this.graphs.remove(graphName, graph);
+ this.eventHub.notify(Events.GRAPH_DROP, g);
+ if (replayAdd) {
try {
- // Delete alias graph in HugeFactory
- HugeFactory.remove((HugeGraph) graph);
- } catch (Exception e) {
- LOG.warn("Failed to remove hugeFactory graph", e);
+ this.loadAddedGraph(graphName);
+ } catch (RuntimeException | Error failure) {
+ // The old DELETE succeeded; a replacement load failure must not invite deleting it again.
+ LOG.error("Failed to load received graph add after cleanup of {}", graphName, failure);
}
}
- this.eventHub.notify(Events.GRAPH_DROP, g);
}
private void checkOptions(String graphSpace, HugeConfig config) {
@@ -2649,41 +2873,66 @@ public final class GraphManager {
continue;
}
LOG.info("Accept graph add signal from etcd for {}", graphName);
- if (this.graphs.containsKey(graphName) ||
- this.creatingGraphs.contains(graphName)) {
- this.creatingGraphs.remove(graphName);
- continue;
+ HugeGraph retryOwner = null;
+ synchronized (this) {
+ HugeGraph owner = (HugeGraph) this.graphs.get(graphName);
+ boolean retryPeerCleanup = owner != null && this.failedPeerRemovals.contains(owner);
+ if (this.creatingGraphs.remove(graphName) && !retryPeerCleanup) {
+ // The acknowledgement of local creation is not a replacement ADD.
+ continue;
+ }
+ if (owner != null) {
+ // Independent key watches can deliver a replacement ADD before the old REMOVE.
+ this.pendingGraphAdds.add(graphName);
+ if (!retryPeerCleanup) {
+ continue;
+ }
+ retryOwner = owner;
+ }
}
-
- LOG.info("Not exist in cache, Starting construct graph {}",
- graphName);
- Map<String, Object> config =
- this.metaManager.getGraphConfig(parts[0], parts[1]);
- if (config == null) {
- LOG.error("The graph config not exist: {}", graphName);
+ if (retryOwner != null) {
+ try {
+ // Retry the exact peer owner; a concurrent replacement must never be closed here.
+ this.dropGraph(parts[0], parts[1], false, retryOwner, true);
+ } catch (HugeException failure) {
+ LOG.error("Failed to retry peer cleanup for graph '{}'", graphName, failure);
+ }
continue;
}
- Object objc = config.get("creator");
- String creator = null == objc ?
- GraphSpace.DEFAULT_CREATOR_NAME :
- String.valueOf(objc);
-
- // Create graph without init
try {
- HugeGraph graph;
- // TODO: add alias graph
- graph = this.createGraph(parts[0], parts[1], creator, config, false);
- LOG.info("Add graph space:{} graph:{}", parts[0], parts[1]);
- graph.started(true);
- if (graph.tx().isOpen()) {
- graph.tx().close();
- }
+ this.loadAddedGraph(graphName);
} catch (HugeException e) {
if (!this.startIgnoreSingleGraphError) {
throw e;
}
- LOG.error(String.format(
- "Failed to create graph '%s'", graphName), e);
+ LOG.error(String.format("Failed to create graph '%s'", graphName), e);
+ }
+ }
+ }
+
+ private void loadAddedGraph(String graphName) {
+ String[] parts = graphName.split(DELIMITER);
+ LOG.info("Not exist in cache, Starting construct graph {}", graphName);
+ Map<String, Object> config = this.metaManager.getGraphConfig(parts[0], parts[1]);
+ HugeGraph graph = null;
+ if (config != null) {
+ Object objc = config.get("creator");
+ String creator = null == objc ? GraphSpace.DEFAULT_CREATOR_NAME : String.valueOf(objc);
+ // Create graph without init; reread metadata instead of keeping an old event's configuration.
+ graph = this.createGraph(parts[0], parts[1], creator, config, false);
+ LOG.info("Add graph space:{} graph:{}", parts[0], parts[1]);
+ graph.started(true);
+ if (graph.tx().isOpen()) {
+ graph.tx().close();
+ }
+ } else {
+ LOG.error("The graph config not exist: {}", graphName);
+ }
+ synchronized (this) {
+ if (this.graphs.get(graphName) == graph &&
+ (graph == null || (!graph.closed() && !this.rpcRemovingOwners.contains(graph)))) {
+ // Do not consume a notification newly retained by another cleanup owner.
+ this.pendingGraphAdds.remove(graphName);
}
}
}
@@ -2704,9 +2953,15 @@ public final class GraphManager {
LOG.error("The graph name format is incorrect: {}", graphName);
continue;
}
+ HugeGraph owner = (HugeGraph) this.graphs.get(graphName);
try {
- this.dropGraph(parts[0], parts[1], false);
+ this.dropGraph(parts[0], parts[1], false, owner, true);
} catch (HugeException e) {
+ synchronized (this) {
+ if (this.graphs.get(graphName) == owner) {
+ this.failedPeerRemovals.add(owner);
+ }
+ }
LOG.error(String.format(
"Failed to drop graph '%s'", graphName), e);
}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java
index 661453e75..745b42c23 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java
@@ -309,6 +309,11 @@ public interface HugeGraph extends Graph, HugeGraphSupplier {
void registerRpcServices(RpcServiceConfig4Server serverConfig,
RpcServiceConfig4Client clientConfig);
+ default void unregisterRpcServices() {
+ // Legacy implementations do not participate in the owned RPC cleanup contract.
+ // Implementations owning graph-scoped RPC resources must override this hook.
+ }
+
default List<String> mapPkId2Name(Collection<Id> ids) {
List<String> names = new ArrayList<>(ids.size());
for (Id id : ids) {
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java
index 11b68f335..51265bfed 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/LoadingMode.java
@@ -153,8 +153,8 @@ public final class LoadingMode {
private void logChange(String action, GraphMode previousMode, int clearedOwners) {
LOG.info("Loading ownership {} for graph '{}' instance {}: mode {} -> {}, epoch {}, " +
- "activeOwners {}, clearedOwners {}", action, this.graphName.get(), this.instance, previousMode, this.mode,
- this.epoch, this.active.size(), clearedOwners);
+ "activeOwners {}, clearedOwners {}", action, this.graphName.get(), this.instance,
+ previousMode, this.mode, this.epoch, this.active.size(), clearedOwners);
}
private void advanceRevision() {
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
index 551d12556..90eb5f070 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
@@ -19,6 +19,7 @@ package org.apache.hugegraph;
import org.apache.hugegraph.exception.HugeException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
@@ -102,6 +103,7 @@ import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.type.define.GraphReadMode;
import org.apache.hugegraph.util.ConfigUtil;
import org.apache.hugegraph.util.DateUtil;
+import org.apache.hugegraph.util.CloseFailure;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Events;
import org.apache.hugegraph.util.LockUtil;
@@ -167,6 +169,8 @@ public class StandardHugeGraph implements HugeGraph {
private final EventHub schemaEventHub;
private final EventHub graphEventHub;
private final EventHub indexEventHub;
+ private final List<Runnable> rpcServiceCleanup = new ArrayList<>();
+ private final List<Runnable> closeResources = new ArrayList<>();
private final LocalCounter localCounter;
private final RateLimiter writeRateLimiter;
private final RateLimiter readRateLimiter;
@@ -177,6 +181,8 @@ public class StandardHugeGraph implements HugeGraph {
private final RamTable ramtable;
private volatile boolean started;
private volatile boolean closed;
+ private boolean closeCompleted;
+ private volatile boolean dropBackendCleared;
private final LoadingMode loadingMode = new LoadingMode(this::spaceGraphName);
private volatile GraphReadMode readMode;
private volatile HugeVariables variables;
@@ -225,6 +231,7 @@ public class StandardHugeGraph implements HugeGraph {
this.started = false;
this.closed = false;
+ this.closeCompleted = false;
this.readMode = GraphReadMode.OLTP_ONLY;
// Init process-wide static configs before lock, so that validation
@@ -429,8 +436,9 @@ public class StandardHugeGraph implements HugeGraph {
if (failure == null) {
throw error;
}
- if (failure != error) {
- failure.addSuppressed(error);
+ Throwable combined = CloseFailure.collect(failure, error);
+ if (combined != failure) {
+ throw (Error) combined;
}
}
}
@@ -1151,7 +1159,7 @@ public class StandardHugeGraph implements HugeGraph {
@Override
public synchronized void close() throws Exception {
- if (this.closed()) {
+ if (this.closed() && this.closeCompleted) {
return;
}
@@ -1165,11 +1173,7 @@ public class StandardHugeGraph implements HugeGraph {
try {
close.run();
} catch (RuntimeException | Error error) {
- if (failure == null) {
- failure = error;
- } else if (failure != error) {
- failure.addSuppressed(error);
- }
+ failure = CloseFailure.collect(failure, error);
}
}
@@ -1182,29 +1186,26 @@ public class StandardHugeGraph implements HugeGraph {
RuntimeException active = new IllegalStateException(String.format(
"Ensure scheduler and tx closed in all threads when closing graph '%s'",
this.spaceGraphName()));
- if (failure == null) {
- failure = active;
- } else {
- failure.addSuppressed(active);
- }
+ failure = CloseFailure.collect(failure, active);
}
if (drained) {
// Plain provider close only notifies listeners; the Raft wrapper also
// stops its context. Owner-close errors do not imply native release.
- this.closed = true;
- for (Runnable close : new Runnable[]{
- () -> CachedGraphTransaction.closeGraph(this.params),
- () -> CachedSchemaTransaction.closeGraph(this.params),
- () -> CachedSchemaTransactionV2.closeGraph(this.params),
- this.storeProvider::close, () -> LockUtil.destroy(this.spaceGraphName())}) {
+ if (!this.closed) {
+ this.closeResources.addAll(Arrays.asList(
+ this::unregisterRpcServices,
+ () -> CachedGraphTransaction.closeGraph(this.params),
+ () -> CachedSchemaTransaction.closeGraph(this.params),
+ () -> CachedSchemaTransactionV2.closeGraph(this.params),
+ this.storeProvider::close, () -> LockUtil.destroy(this.spaceGraphName())));
+ this.closed = true;
+ }
+ for (Iterator<Runnable> closes = this.closeResources.iterator(); closes.hasNext();) {
try {
- close.run();
+ closes.next().run();
+ closes.remove();
} catch (RuntimeException | Error error) {
- if (failure == null) {
- failure = error;
- } else if (failure != error) {
- failure.addSuppressed(error);
- }
+ failure = CloseFailure.collect(failure, error);
}
}
}
@@ -1214,6 +1215,8 @@ public class StandardHugeGraph implements HugeGraph {
if (failure != null) {
throw (RuntimeException) failure;
}
+ // Logical closure fences access before resources close; only success ends cleanup retries.
+ this.closeCompleted = true;
}
@Override
@@ -1229,39 +1232,35 @@ public class StandardHugeGraph implements HugeGraph {
@Override
public void drop() {
- this.checkGraphNotClosed();
- TaskScheduler scheduler = this.taskManager.getScheduler(this.params);
- if (scheduler != null && !scheduler.close()) {
- // Fence admission before waiting for actual task owners to exit.
- try {
- scheduler.waitUntilAllTasksCompleted(this.configuration().get(CoreOptions.TASK_WAIT_TIMEOUT));
- } catch (TimeoutException e) {
- throw new HugeException("Tasks are still active in graph '%s'; task admission remains closed. " +
- "Retry deletion after owners exit, or restart to resume task admission",
- e, this.spaceGraphName());
+ if (!this.dropBackendCleared) {
+ this.checkGraphNotClosed();
+ TaskScheduler scheduler = this.taskManager.getScheduler(this.params);
+ if (scheduler != null && !scheduler.close()) {
+ // Fence admission before waiting for actual task owners to exit.
+ try {
+ scheduler.waitUntilAllTasksCompleted(this.configuration().get(CoreOptions.TASK_WAIT_TIMEOUT));
+ } catch (TimeoutException e) {
+ throw new HugeException("Tasks are still active in graph '%s'; task admission remains closed. " +
+ "Retry deletion after owners exit, or restart to resume task admission",
+ e, this.spaceGraphName());
+ }
+ E.checkState(scheduler.close(),
+ "Tasks are still active in graph '%s'; task admission remains closed. " +
+ "Retry deletion after owners exit, or restart to resume task admission",
+ this.spaceGraphName());
}
- E.checkState(scheduler.close(),
- "Tasks are still active in graph '%s'; task admission remains closed. " +
- "Retry deletion after owners exit, or restart to resume task admission",
- this.spaceGraphName());
+ this.clearBackend();
+ // Only this same owner's successful drop clear permits cleanup after logical close.
+ this.dropBackendCleared = true;
}
- this.clearBackend();
try {
- /*
- * It's hard to ensure all threads close the tx.
- * TODO:
- * - schedule a tx-close to each thread,
- * or
- * - add forceClose() method to backend store.
- */
this.close();
- } catch (Throwable e) {
- if (!this.closed()) {
- throw new HugeException("Failed to close graph '%s', please retry later",
- e, this.spaceGraphName());
- }
- LOG.warn("Failed to close graph {} {}", this, e);
+ } catch (Error failure) {
+ throw failure;
+ } catch (Exception failure) {
+ throw new HugeException("Failed to close graph '%s', please retry later",
+ failure, this.spaceGraphName());
}
HugeConfig config = this.configuration();
this.storeProvider.onDeleteConfig(config);
@@ -1395,26 +1394,85 @@ public class StandardHugeGraph implements HugeGraph {
}
@Override
- public void registerRpcServices(RpcServiceConfig4Server serverConfig,
- RpcServiceConfig4Client clientConfig) {
+ public synchronized void registerRpcServices(RpcServiceConfig4Server serverConfig,
+ RpcServiceConfig4Client clientConfig) {
/*
* Skip register cache-rpc service if it's non-shared storage,
* because we assume cache of non-shared storage is updated by raft.
*/
- if (!this.backendStoreFeatures().supportsSharedStorage()) {
+ if (!this.rpcServiceCleanup.isEmpty()) {
return;
}
+ boolean opened = this.tx.isOpen();
+ boolean shared;
+ Throwable readFailure = null;
+ try {
+ shared = this.backendStoreFeatures().supportsSharedStorage();
+ } catch (RuntimeException | Error error) {
+ readFailure = error;
+ throw error;
+ } finally {
+ if (!opened && this.tx.isOpen()) {
+ try {
+ this.closeCurrentThreadTransaction();
+ } catch (RuntimeException | Error cleanup) {
+ if (readFailure == null) {
+ throw cleanup;
+ }
+ if (readFailure != cleanup) {
+ readFailure.addSuppressed(cleanup);
+ }
+ }
+ }
+ }
+ if (!shared) {
+ return;
+ }
+ E.checkState(!this.closed(), "Can't register RPC services for a closed graph");
+ String graph = this.spaceGraphName();
+ try {
+ GraphCacheNotifier graphProxy = clientConfig.serviceProxy(graph, GraphCacheNotifier.class);
+ this.rpcServiceCleanup.add(() -> clientConfig.removeServiceProxy(
+ graph, GraphCacheNotifier.class.getName(), graphProxy));
+ HugeGraphCacheNotifier graphNotifier = new HugeGraphCacheNotifier(this.graphEventHub, graphProxy);
+ this.rpcServiceCleanup.add(graphNotifier::close);
+ String graphService = serverConfig.addService(graph, GraphCacheNotifier.class, graphNotifier);
+ this.rpcServiceCleanup.add(() -> serverConfig.removeService(graphService, graphNotifier));
+
+ SchemaCacheNotifier schemaProxy = clientConfig.serviceProxy(graph, SchemaCacheNotifier.class);
+ this.rpcServiceCleanup.add(() -> clientConfig.removeServiceProxy(
+ graph, SchemaCacheNotifier.class.getName(), schemaProxy));
+ HugeSchemaCacheNotifier schemaNotifier = new HugeSchemaCacheNotifier(this.schemaEventHub, schemaProxy);
+ this.rpcServiceCleanup.add(schemaNotifier::close);
+ String schemaService = serverConfig.addService(graph, SchemaCacheNotifier.class, schemaNotifier);
+ this.rpcServiceCleanup.add(() -> serverConfig.removeService(schemaService, schemaNotifier));
+ } catch (RuntimeException | Error failure) {
+ try {
+ this.unregisterRpcServices();
+ } catch (RuntimeException | Error cleanup) {
+ failure.addSuppressed(cleanup);
+ }
+ throw failure;
+ }
+ }
- Class<GraphCacheNotifier> clazz1 = GraphCacheNotifier.class;
- // The proxy is sometimes unavailable (issue #664)
- CacheNotifier proxy = clientConfig.serviceProxy(this.spaceGraphName(), clazz1);
- serverConfig.addService(this.spaceGraphName(), clazz1, new HugeGraphCacheNotifier(
- this.graphEventHub, proxy));
-
- Class<SchemaCacheNotifier> clazz2 = SchemaCacheNotifier.class;
- proxy = clientConfig.serviceProxy(this.spaceGraphName(), clazz2);
- serverConfig.addService(this.spaceGraphName(), clazz2, new HugeSchemaCacheNotifier(
- this.schemaEventHub, proxy));
+ @Override
+ public synchronized void unregisterRpcServices() {
+ Throwable failure = null;
+ for (int i = this.rpcServiceCleanup.size() - 1; i >= 0; i--) {
+ try {
+ this.rpcServiceCleanup.get(i).run();
+ this.rpcServiceCleanup.remove(i);
+ } catch (RuntimeException | Error error) {
+ failure = CloseFailure.collect(failure, error);
+ }
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure != null) {
+ throw (RuntimeException) failure;
+ }
}
private void waitUntilAllTasksCompleted() {
@@ -1458,11 +1516,7 @@ public class StandardHugeGraph implements HugeGraph {
try {
close.run();
} catch (RuntimeException | Error error) {
- if (failure == null) {
- failure = error;
- } else if (failure != error) {
- failure.addSuppressed(error);
- }
+ failure = CloseFailure.collect(failure, error);
}
}
if (failure instanceof Error) {
@@ -1502,10 +1556,14 @@ public class StandardHugeGraph implements HugeGraph {
private final EventHub hub;
private final EventListener cacheEventListener;
+ private volatile boolean closed;
public AbstractCacheNotifier(EventHub hub, CacheNotifier proxy) {
this.hub = hub;
this.cacheEventListener = event -> {
+ if (this.closed) {
+ return false;
+ }
try {
LOG.info("Received event: {}", event);
Object[] args = event.args();
@@ -1549,6 +1607,7 @@ public class StandardHugeGraph implements HugeGraph {
@Override
public void close() {
+ this.closed = true;
this.hub.unlisten(Events.CACHE, this.cacheEventListener);
}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftBackendStoreProvider.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftBackendStoreProvider.java
index ff6cfee8c..b704efdaf 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftBackendStoreProvider.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftBackendStoreProvider.java
@@ -29,6 +29,7 @@ import org.apache.hugegraph.backend.store.raft.rpc.RaftRequests.StoreType;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.event.EventHub;
import org.apache.hugegraph.event.EventListener;
+import org.apache.hugegraph.util.CloseFailure;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Events;
import org.apache.hugegraph.util.Log;
@@ -44,6 +45,9 @@ public class RaftBackendStoreProvider implements BackendStoreProvider {
private final BackendStoreProvider provider;
private final RaftContext context;
+ private boolean providerClosed;
+ private boolean contextClosed;
+
private RaftBackendStore schemaStore;
private RaftBackendStore graphStore;
private RaftBackendStore systemStore;
@@ -158,10 +162,29 @@ public class RaftBackendStoreProvider implements BackendStoreProvider {
}
@Override
- public void close() {
- this.provider.close();
- if (this.context != null) {
- this.context.close();
+ public synchronized void close() {
+ Throwable failure = null;
+ try {
+ if (!this.providerClosed) {
+ this.provider.close();
+ this.providerClosed = true;
+ }
+ } catch (RuntimeException | Error error) {
+ failure = error;
+ }
+ try {
+ if (!this.contextClosed && this.context != null) {
+ this.context.close();
+ this.contextClosed = true;
+ }
+ } catch (RuntimeException | Error error) {
+ failure = CloseFailure.collect(failure, error);
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure != null) {
+ throw (RuntimeException) failure;
}
}
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftContext.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftContext.java
index ea2e92bfc..5545da83b 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftContext.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftContext.java
@@ -28,6 +28,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.hugegraph.exception.HugeException;
@@ -53,6 +54,7 @@ import org.apache.hugegraph.event.EventHub;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.util.Bytes;
+import org.apache.hugegraph.util.CloseFailure;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Events;
import org.apache.hugegraph.util.Log;
@@ -102,6 +104,9 @@ public final class RaftContext {
private final ExecutorService backendExecutor;
private RpcServer raftRpcServer;
+ private boolean nodeClosed;
+ private boolean rpcClosed;
+ private boolean sharedRpcServer;
private PeerId endpoint;
private RaftNode raftNode;
@@ -169,15 +174,61 @@ public final class RaftContext {
node.waitRaftLogSynced(RaftContext.NO_TIMEOUT);
}
- public void close() {
+ public synchronized void close() {
LOG.info("Stop raft server: {}", this.endpoint());
- RaftNode node = this.node();
- if (node != null) {
- node.shutdown();
+ Throwable failure = null;
+ try {
+ RaftNode node = this.node();
+ if (!this.nodeClosed) {
+ if (node != null) {
+ node.shutdown();
+ }
+ this.nodeClosed = true;
+ }
+ } catch (RuntimeException | Error error) {
+ failure = error;
+ }
+ try {
+ this.shutdownRpcServer();
+ } catch (RuntimeException | Error error) {
+ failure = CloseFailure.collect(failure, error);
+ }
+ // Only a joined node has stopped producing work for these per-context executors.
+ if (this.nodeClosed) {
+ for (ExecutorService executor : new ExecutorService[]{this.readIndexExecutor,
+ this.snapshotExecutor,
+ this.backendExecutor}) {
+ try {
+ this.closeExecutor(executor);
+ } catch (RuntimeException | Error error) {
+ failure = CloseFailure.collect(failure, error);
+ }
+ }
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure != null) {
+ throw (RuntimeException) failure;
}
+ }
- this.shutdownRpcServer();
+ private void closeExecutor(ExecutorService executor) {
+ if (executor == null || executor.isTerminated()) {
+ return;
+ }
+ // Keep queued work: shutdownNow() would discard pending apply/snapshot operations.
+ executor.shutdown();
+ try {
+ long timeout = this.config().get(CoreOptions.TASK_WAIT_TIMEOUT);
+ if (!executor.awaitTermination(timeout, TimeUnit.SECONDS)) {
+ throw new RaftException("Raft executor is still active, please retry close later");
+ }
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new RaftException("Interrupted while draining raft executor", interrupted);
+ }
}
public RaftNode node() {
@@ -372,6 +423,7 @@ public final class RaftContext {
@SuppressWarnings("unused")
private RpcServer initAndStartRpcServer() {
+ this.sharedRpcServer = false;
Integer lowWaterMark = this.config().get(
CoreOptions.RAFT_RPC_BUF_LOW_WATER_MARK);
System.setProperty("bolt.channel_write_buf_low_water_mark",
@@ -390,6 +442,8 @@ public final class RaftContext {
}
private RpcServer wrapRpcServer(com.alipay.remoting.rpc.RpcServer rpcServer) {
+ // GraphManager owns this listener, which is shared by all local graph contexts.
+ this.sharedRpcServer = true;
// TODO: pass ServerOptions instead of CoreOptions, to share by graphs
Integer lowWaterMark = this.config().get(
CoreOptions.RAFT_RPC_BUF_LOW_WATER_MARK);
@@ -408,9 +462,17 @@ public final class RaftContext {
}
private void shutdownRpcServer() {
- this.raftRpcServer.shutdown();
+ if (this.sharedRpcServer) {
+ return;
+ }
+ if (!this.rpcClosed && this.raftRpcServer != null) {
+ this.raftRpcServer.shutdown();
+ this.rpcClosed = true;
+ }
PeerId endpoint = this.endpoint();
- NodeManager.getInstance().removeAddress(endpoint.getEndpoint());
+ if (endpoint != null) {
+ NodeManager.getInstance().removeAddress(endpoint.getEndpoint());
+ }
}
private void registerRpcRequestProcessors() {
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftNode.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftNode.java
index 3b39ea3de..956fd1859 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftNode.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/raft/RaftNode.java
@@ -99,15 +99,15 @@ public final class RaftNode {
LOG.info("Shutdown raft node: {}", this);
this.node.shutdown();
- if (this.raftGroupService != null) {
- this.raftGroupService.shutdown();
- try {
- this.raftGroupService.join();
- } catch (final InterruptedException e) {
- throw new RaftException(
- "Interrupted while shutdown raftGroupService");
- }
+ // RaftGroupService.shutdown() also removes the shared server address.
+ // Close only this node; GraphManager owns the shared listener and address.
+ try {
+ this.node.join();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RaftException("Interrupted while shutting down raft node", e);
}
+ this.raftGroupService = null;
}
public RaftClosure<?> snapshot() {
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java
index ced7e1342..4e050c33e 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java
@@ -553,13 +553,17 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler {
HugeTask<?> running = this.runningTasks.get(id);
if (running != null) {
- this.cancelOwned(this.executions.get(id), () -> {
+ boolean cancelled = this.cancelOwned(this.executions.get(id), () -> {
this.markTaskDeleting(running);
return running.cancel(true);
}, false);
- @SuppressWarnings("unchecked")
- HugeTask<V> result = (HugeTask<V>) running;
- return result;
+ if (cancelled || this.executions.containsKey(id)) {
+ @SuppressWarnings("unchecked")
+ HugeTask<V> result = (HugeTask<V>) running;
+ return result;
+ }
+ // The local owner finished before cancellation; delete its now-completed database record.
+ task = this.taskWithoutResult(id);
}
if (!force && !task.completed()) {
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java
index b53742a54..841e8bdb7 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java
@@ -428,6 +428,12 @@ public class StandardTaskScheduler implements TaskScheduler {
task.scheduler(this);
E.checkArgumentNotNull(task, "Task can't be null");
this.call(() -> {
+ // Completion can save after deletion, while this exact
+ // execution owner is still retained for native cleanup.
+ if (task.status() == TaskStatus.DELETING && this.tasks.get(task.id()) == task &&
+ QueryResults.one(this.tx().queryTaskInfos(task.id())) == null) {
+ return null;
+ }
// Construct vertex from task
HugeVertex vertex = this.tx().constructVertex(task);
// Delete index of old vertex to avoid stale index
@@ -509,7 +515,7 @@ public class StandardTaskScheduler implements TaskScheduler {
E.checkArgumentNotNull(id, "Parameter task id can't be null");
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
- if (task != null) {
+ if (task != null && task.status() != TaskStatus.DELETING) {
return withResult ? task : task.copyWithoutResult();
}
return this.findTask(id, withResult);
@@ -528,7 +534,7 @@ public class StandardTaskScheduler implements TaskScheduler {
for (Id id : ids) {
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
- if (task != null) {
+ if (task != null && task.status() != TaskStatus.DELETING) {
taskInMem.add(withResult ? task : task.copyWithoutResult());
} else {
taskIdsNotInMem.add(id);
@@ -625,13 +631,17 @@ public class StandardTaskScheduler implements TaskScheduler {
* when the database status is inconsistent.
*/
if (running != null) {
- E.checkArgument(force || running.completed(),
+ E.checkArgument(force || running.completed() ||
+ (running.status() == TaskStatus.DELETING && task.completed()),
"Can't delete incomplete task '%s' in status %s" +
", Please try to cancel the task first",
id, running.status());
- if (force && !running.completed()) {
+ boolean completed = running.completed();
+ if (force) {
running.overwriteStatus(TaskStatus.DELETING);
- running.cancel(true);
+ if (!completed) {
+ running.cancel(true);
+ }
}
this.remove(running, force);
} else if (task != null) {
@@ -652,6 +662,11 @@ public class StandardTaskScheduler implements TaskScheduler {
E.checkState(force || result.completed(),
"Can't delete incomplete task '%s' in status %s",
id, result.status());
+ // A non-force request must pass the persisted completion check
+ // before its retained owner can be marked for deletion.
+ if (!force && running != null && this.tasks.get(id) == running) {
+ running.overwriteStatus(TaskStatus.DELETING);
+ }
this.tx().removeTaskVertex(vertex);
return result;
});
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/CloseFailure.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/CloseFailure.java
new file mode 100644
index 000000000..9a35a6e35
--- /dev/null
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/CloseFailure.java
@@ -0,0 +1,35 @@
+/*
+ * 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.util;
+
+public final class CloseFailure {
+
+ private CloseFailure() {
+ }
+
+ public static Throwable collect(Throwable first, Throwable next) {
+ if (first == null || first == next) {
+ return first == null ? next : first;
+ }
+ Throwable primary = next instanceof Error && !(first instanceof Error) ? next : first;
+ Throwable secondary = primary == first ? next : first;
+ primary.addSuppressed(secondary);
+ return primary;
+ }
+
+}
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java
index 6ab384419..244910773 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java
@@ -558,6 +558,64 @@ public class DistributedTaskAdmissionTest {
}
}
+ @Test
+ public void testDeleteAfterLocalOwnerFinishesStillRemovesRecord() throws Exception {
+ try (Fixture fixture = new Fixture("delete_finished_owner")) {
+ DeleteCompletionCallable.started = new CountDownLatch(1);
+ DeleteCompletionCallable.finish = new CountDownLatch(1);
+ HugeTask<?> task = new HugeTask<>(IdGenerator.of(9999952L), null, new DeleteCompletionCallable());
+ task.type("test");
+ task.name("delete-finished-owner");
+ Map<Id, HugeTask<?>> running = Whitebox.getInternalState(fixture.scheduler, "runningTasks");
+ Map<Id, HugeTask<?>> observed = Mockito.spy(running);
+ AtomicBoolean intercept = new AtomicBoolean();
+ Mockito.doAnswer(invocation -> {
+ HugeTask<?> value = (HugeTask<?>) invocation.callRealMethod();
+ if (intercept.compareAndSet(true, false)) {
+ Assert.assertSame(task, value);
+ DeleteCompletionCallable.finish.countDown();
+ // The actual worker completes persistence and releases ownership before delete continues.
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ Assert.assertEquals(TaskStatus.SUCCESS, task.status());
+ Assert.assertEquals(TaskStatus.SUCCESS, fixture.scheduler.task(task.id(), false).status());
+ Assert.assertEquals(0, fixture.scheduler.pendingTasks());
+ }
+ return value;
+ }).when(observed).get(task.id());
+ Whitebox.setInternalState(fixture.scheduler, "runningTasks", observed);
+ try {
+ fixture.scheduler.schedule(task);
+ Assert.assertTrue(DeleteCompletionCallable.started.await(10L, TimeUnit.SECONDS));
+ intercept.set(true);
+ Assert.assertNotNull(fixture.scheduler.delete(task.id(), false));
+ Assert.assertFalse(intercept.get());
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> fixture.scheduler.task(task.id(), false));
+ fixture.assertWorkerClean();
+ } finally {
+ DeleteCompletionCallable.finish.countDown();
+ }
+ }
+ }
+
+ public static class DeleteCompletionCallable extends org.apache.hugegraph.job.UserJob<Object> {
+
+ private static CountDownLatch started;
+ private static CountDownLatch finish;
+
+ @Override
+ public String type() {
+ return "test";
+ }
+
+ @Override
+ public Object execute() {
+ started.countDown();
+ awaitUninterruptibly(finish);
+ return null;
+ }
+ }
+
private static void awaitUninterruptibly(CountDownLatch release) {
while (release.getCount() != 0L) {
try {
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java
index 78a8e66b0..0248e7cd2 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java
@@ -392,6 +392,463 @@ public class StandardTaskAdmissionTest {
}
}
+ @Test
+ public void testForceDeleteHidesPublicTaskWhileNativeWorkerRetainsOwnership() throws Exception {
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch interrupted = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ java.util.Map<String, Object> observed = new java.util.LinkedHashMap<>();
+ boolean publicPresent;
+ boolean listPresent;
+ try (Fixture fixture = new Fixture("standard_force_delete_native")) {
+ ExecutorService observerWorker = Executors.newSingleThreadExecutor();
+ ExecutorService observerDatabase = Executors.newSingleThreadExecutor(runnable ->
+ new Thread(runnable, "standard-independent-db"));
+ StandardTaskScheduler observer = new StandardTaskScheduler(fixture.params, observerWorker,
+ observerDatabase);
+ BlockingJob job = new BlockingJob() {
+ @Override
+ public Object execute() {
+ this.graph().vertices().hasNext();
+ started.countDown();
+ boolean wasInterrupted = false;
+ while (release.getCount() != 0L) {
+ try {
+ release.await();
+ } catch (InterruptedException ignored) {
+ wasInterrupted = true;
+ interrupted.countDown();
+ }
+ }
+ if (wasInterrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return null;
+ }
+ };
+ HugeTask<Object> task = fixture.task(job);
+ try {
+ fixture.scheduler.save(task);
+ Assert.assertNotNull(observer.findTask(task.id()));
+ fixture.scheduler.schedule(task);
+ Assert.assertTrue(started.await(10L, TimeUnit.SECONDS));
+ Assert.assertNotNull(observer.findTask(task.id()));
+ HugeTask<?> deleted = fixture.scheduler.delete(task.id(), true);
+ Assert.assertNotNull(deleted);
+ Assert.assertTrue(interrupted.await(10L, TimeUnit.SECONDS));
+ Assert.assertEquals(1L, release.getCount());
+ try {
+ observed.put("public_task_status_before_release",
+ fixture.scheduler.task(task.id()).status().name());
+ publicPresent = true;
+ } catch (org.apache.hugegraph.exception.NotFoundException missing) {
+ publicPresent = false;
+ }
+ listPresent = fixture.scheduler.tasks(java.util.Collections.singletonList(task.id())).hasNext();
+ try {
+ fixture.scheduler.task(task.id(), false);
+ observed.put("public_task_without_result_present_before_release", true);
+ } catch (org.apache.hugegraph.exception.NotFoundException missing) {
+ observed.put("public_task_without_result_present_before_release", false);
+ }
+ observed.put("public_ids_without_result_present_before_release", fixture.scheduler.tasks(
+ java.util.Collections.singletonList(task.id()), false).hasNext());
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> observer.findTask(task.id()));
+ Assert.assertFalse(observer.findAllTask(10L, null, false).hasNext());
+ int pending = fixture.scheduler.pendingTasks();
+ boolean closed = fixture.scheduler.close();
+ observed.put("public_task_present_before_release", publicPresent);
+ observed.put("public_task_ids_list_present_before_release", listPresent);
+ observed.put("independent_committed_db_present_before_release", false);
+ observed.put("native_worker_interrupted_and_still_held", true);
+ observed.put("pending_before_release", pending);
+ observed.put("close_before_release", closed);
+ observed.put("logical_task_cancelled", task.isCancelled());
+ observed.put("native_scheduler_origin", StandardTaskScheduler.class.getProtectionDomain()
+ .getCodeSource().getLocation().toString());
+ Assert.assertEquals(1, pending);
+ Assert.assertFalse(closed);
+ release.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ fixture.scheduler.waitUntilAllTasksCompleted(10L);
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> fixture.scheduler.task(task.id()));
+ Assert.assertFalse(fixture.scheduler.tasks(java.util.Collections.singletonList(task.id())).hasNext());
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> observer.findTask(task.id()));
+ Assert.assertEquals(0, fixture.scheduler.pendingTasks());
+ Assert.assertTrue(fixture.scheduler.close());
+ observed.put("public_task_present_after_release", false);
+ observed.put("public_task_ids_list_present_after_release", false);
+ observed.put("independent_committed_db_present_after_release", false);
+ observed.put("pending_after_release", 0);
+ observed.put("close_after_release", true);
+ observed.put("one_persisted_task_one_delete", true);
+ String evidence = System.getProperty("hugegraph.task.delete.evidence");
+ if (evidence != null) {
+ java.nio.file.Files.writeString(java.nio.file.Paths.get(evidence, "standard-observation.json"),
+ org.apache.hugegraph.util.JsonUtil.toJson(observed));
+ }
+ } finally {
+ release.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ Assert.assertTrue(observer.close());
+ observerWorker.shutdownNow();
+ observerDatabase.shutdownNow();
+ Assert.assertTrue(observerWorker.awaitTermination(10L, TimeUnit.SECONDS));
+ Assert.assertTrue(observerDatabase.awaitTermination(10L, TimeUnit.SECONDS));
+ }
+ }
+ // Admission accounting must keep the true native owner above, while
+ // the established Standard force-delete API contract requires absence.
+ Assert.assertFalse("force-deleted task is still publicly visible", publicPresent);
+ Assert.assertFalse("force-deleted task is still listed by ID", listPresent);
+ Assert.assertEquals(false, observed.get("public_task_without_result_present_before_release"));
+ Assert.assertEquals(false, observed.get("public_ids_without_result_present_before_release"));
+ }
+
+ @Test
+ public void testDeletingTaskStillVisibleWhileItsRecordExists() throws Exception {
+ try (Fixture fixture = new Fixture("standard_deleting_db_visible")) {
+ BlockingJob job = new BlockingJob();
+ HugeTask<Object> task = fixture.task(job);
+ fixture.scheduler.schedule(task);
+ try {
+ Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS));
+ task.overwriteStatus(TaskStatus.DELETING);
+ fixture.scheduler.save(task);
+ for (boolean withResult : new boolean[]{true, false}) {
+ Assert.assertEquals(TaskStatus.DELETING, fixture.scheduler.task(task.id(), withResult).status());
+ java.util.Iterator<HugeTask<Object>> tasks = fixture.scheduler.tasks(
+ java.util.Collections.singletonList(task.id()), withResult);
+ Assert.assertTrue(tasks.hasNext());
+ Assert.assertEquals(TaskStatus.DELETING, tasks.next().status());
+ Assert.assertFalse(tasks.hasNext());
+ }
+ } finally {
+ job.release.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ }
+ }
+ }
+
+ @Test
+ public void testOrdinaryTaskVisibilityRetainsResultSelection() throws Exception {
+ try (Fixture fixture = new Fixture("standard_normal_task_visible")) {
+ HugeTask<Object> task = fixture.task(new ResultJob());
+ fixture.scheduler.schedule(task).get(10L, TimeUnit.SECONDS);
+ for (boolean withResult : new boolean[]{true, false}) {
+ HugeTask<Object> loaded = fixture.scheduler.task(task.id(), withResult);
+ Assert.assertEquals(TaskStatus.SUCCESS, loaded.status());
+ Assert.assertEquals(withResult ? task.result() : null, loaded.result());
+ java.util.Iterator<HugeTask<Object>> tasks = fixture.scheduler.tasks(
+ java.util.Collections.singletonList(task.id()), withResult);
+ Assert.assertTrue(tasks.hasNext());
+ HugeTask<Object> listed = tasks.next();
+ Assert.assertEquals(TaskStatus.SUCCESS, listed.status());
+ Assert.assertEquals(withResult ? task.result() : null, listed.result());
+ Assert.assertFalse(tasks.hasNext());
+ }
+ }
+ }
+
+ @Test
+ public void testForceDeleteCompletedOwnerDoesNotResurrectAfterDoneSave() throws Exception {
+ java.util.Map<String, Object> observed = new java.util.LinkedHashMap<>();
+ boolean publicPresent;
+ boolean idsPresent;
+ boolean dbResurrected;
+ try (Fixture fixture = new Fixture("standard_force_delete_done_save")) {
+ HugeGraphParams nativeParams = (HugeGraphParams) Whitebox.getInternalState(fixture.graph, "params");
+ @SuppressWarnings("unchecked")
+ java.util.Map<HugeGraphParams, org.apache.hugegraph.task.TaskScheduler> schedulers =
+ (java.util.Map<HugeGraphParams, org.apache.hugegraph.task.TaskScheduler>) Whitebox
+ .getInternalState(org.apache.hugegraph.task.TaskManager.instance(), "schedulers");
+ org.apache.hugegraph.task.TaskScheduler original = schedulers.put(nativeParams, fixture.scheduler);
+ ExecutorService observerWorker = Executors.newSingleThreadExecutor();
+ ExecutorService observerDatabase = Executors.newSingleThreadExecutor();
+ StandardTaskScheduler observer = new StandardTaskScheduler(fixture.params, observerWorker,
+ observerDatabase);
+ CompletionSaveJob job = new CompletionSaveJob();
+ HugeTask<Object> task = fixture.task(job);
+ try {
+ Assert.assertSame(fixture.scheduler, fixture.graph.taskScheduler());
+ fixture.scheduler.schedule(task);
+ Assert.assertTrue(job.beforeSave.await(10L, TimeUnit.SECONDS));
+ Assert.assertEquals(TaskStatus.SUCCESS, task.status());
+ Assert.assertTrue(task.completed());
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ Assert.assertNotNull(observer.findTask(task.id()));
+ HugeTask<?> deleted = fixture.scheduler.delete(task.id(), true);
+ Assert.assertNotNull(deleted);
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> observer.findTask(task.id()));
+ try {
+ observed.put("public_status_after_delete_before_done_save",
+ fixture.scheduler.task(task.id()).status().name());
+ publicPresent = true;
+ } catch (org.apache.hugegraph.exception.NotFoundException missing) {
+ publicPresent = false;
+ }
+ idsPresent = fixture.scheduler.tasks(java.util.Collections.singletonList(task.id())).hasNext();
+ observed.put("success_owner_retained_before_delete", true);
+ observed.put("public_present_after_delete_before_done_save", publicPresent);
+ observed.put("ids_present_after_delete_before_done_save", idsPresent);
+ observed.put("committed_db_present_after_delete_before_done_save", false);
+ observed.put("pending_before_done_save", fixture.scheduler.pendingTasks());
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ Assert.assertFalse(fixture.scheduler.close());
+ observed.put("close_before_done_save", false);
+ job.releaseSave.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ Assert.assertTrue(job.saved.await(10L, TimeUnit.SECONDS));
+ try {
+ observed.put("db_status_after_real_done_save", observer.findTask(task.id()).status().name());
+ dbResurrected = true;
+ } catch (org.apache.hugegraph.exception.NotFoundException missing) {
+ dbResurrected = false;
+ }
+ observed.put("committed_db_resurrected_after_real_done_save", dbResurrected);
+ observed.put("pending_after_real_done_save", fixture.scheduler.pendingTasks());
+ Assert.assertEquals(0, fixture.scheduler.pendingTasks());
+ Assert.assertTrue(fixture.scheduler.close());
+ observed.put("close_after_real_done_save", true);
+ observed.put("real_sysjob_done_save_scheduler_binding", true);
+ String evidence = System.getProperty("hugegraph.task.delete.evidence");
+ if (evidence != null) {
+ java.nio.file.Files.writeString(
+ java.nio.file.Paths.get(evidence, "completion-save-observation.json"),
+ org.apache.hugegraph.util.JsonUtil.toJson(observed));
+ }
+ } finally {
+ job.releaseSave.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ Assert.assertTrue(schedulers.replace(nativeParams, fixture.scheduler, original));
+ Assert.assertTrue(observer.close());
+ observerWorker.shutdownNow();
+ observerDatabase.shutdownNow();
+ Assert.assertTrue(observerWorker.awaitTermination(10L, TimeUnit.SECONDS));
+ Assert.assertTrue(observerDatabase.awaitTermination(10L, TimeUnit.SECONDS));
+ }
+ }
+ Assert.assertFalse("completed owner still visible after force deletion", publicPresent);
+ Assert.assertFalse("completed owner still listed after force deletion", idsPresent);
+ Assert.assertFalse("real done save recreated force-deleted task", dbResurrected);
+ }
+
+ @Test
+ public void testDoneSaveBeforeDeleteRetainsNativeCleanupOwner() throws Exception {
+ this.doneSaveBeforeDelete(true);
+ }
+
+ @Test
+ public void testNonForceCompletedDeleteRetainsNativeCleanupOwner() throws Exception {
+ this.doneSaveBeforeDelete(false);
+ }
+
+ @Test
+ public void testNonForceDeleteDatabaseFailureCanRetryWithDurableCompletion() throws Exception {
+ this.doneSaveBeforeDelete(false, true);
+ }
+
+ private void doneSaveBeforeDelete(boolean force) throws Exception {
+ this.doneSaveBeforeDelete(force, false);
+ }
+
+ private void doneSaveBeforeDelete(boolean force, boolean failure) throws Exception {
+ try (Fixture fixture = new Fixture("standard_save_before_delete");
+ SchedulerBinding binding = new SchedulerBinding(fixture)) {
+ CountDownLatch closing = new CountDownLatch(1);
+ CountDownLatch close = new CountDownLatch(1);
+ AtomicBoolean first = new AtomicBoolean(true);
+ Mockito.doAnswer(invocation -> {
+ if (Thread.currentThread().getName().equals("standard-test-worker") &&
+ first.compareAndSet(true, false)) {
+ closing.countDown();
+ await(close);
+ }
+ return invocation.callRealMethod();
+ }).when(fixture.params).closeTx();
+ HugeTask<Object> task = fixture.task(new ResultJob());
+ try {
+ fixture.scheduler.schedule(task);
+ Assert.assertTrue(closing.await(10L, TimeUnit.SECONDS));
+ Assert.assertEquals(TaskStatus.SUCCESS, fixture.scheduler.findTask(task.id()).status());
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ if (failure) {
+ org.apache.hugegraph.task.TaskTransaction transaction = (org.apache.hugegraph.task.TaskTransaction)
+ Whitebox.getInternalState(fixture.scheduler, "taskTx");
+ org.apache.hugegraph.task.TaskTransaction rejecting = Mockito.spy(transaction);
+ Error rejected = new AssertionError("native completed delete rejected before mutation");
+ Mockito.doThrow(rejected).doCallRealMethod().when(rejecting)
+ .removeTaskVertex(Mockito.any(org.apache.hugegraph.structure.HugeVertex.class));
+ Whitebox.setInternalState(fixture.scheduler, "taskTx", rejecting);
+ Throwable failed = Assert.assertThrows(RuntimeException.class,
+ () -> fixture.scheduler.delete(task.id(), force));
+ Throwable cause = failed;
+ while (cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ Assert.assertSame(rejected, cause);
+ Assert.assertEquals(TaskStatus.SUCCESS, fixture.scheduler.findTask(task.id()).status());
+ Assert.assertEquals(TaskStatus.SUCCESS, fixture.scheduler.task(task.id()).status());
+ Assert.assertEquals(TaskStatus.DELETING, task.status());
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ }
+ fixture.scheduler.delete(task.id(), force);
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> fixture.scheduler.findTask(task.id()));
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ Assert.assertFalse(fixture.scheduler.close());
+ assertTaskAbsent(fixture.scheduler, task);
+ } finally {
+ close.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ }
+ assertTaskAbsent(fixture.scheduler, task);
+ Assert.assertEquals(0, fixture.scheduler.pendingTasks());
+ }
+ }
+
+ @Test
+ public void testDeleteDatabaseFailureRetainsRecordForRetryAndOriginalError() throws Exception {
+ try (Fixture fixture = new Fixture("standard_delete_db_failure");
+ SchedulerBinding binding = new SchedulerBinding(fixture)) {
+ CompletionSaveJob job = new CompletionSaveJob();
+ HugeTask<Object> task = fixture.task(job);
+ try {
+ fixture.scheduler.schedule(task);
+ Assert.assertTrue(job.beforeSave.await(10L, TimeUnit.SECONDS));
+ org.apache.hugegraph.task.TaskTransaction transaction = (org.apache.hugegraph.task.TaskTransaction)
+ Whitebox.getInternalState(fixture.scheduler, "taskTx");
+ org.apache.hugegraph.task.TaskTransaction rejecting = Mockito.spy(transaction);
+ Error rejected = new AssertionError("native delete boundary rejected before mutation");
+ Mockito.doThrow(rejected).doCallRealMethod().when(rejecting)
+ .removeTaskVertex(Mockito.any(org.apache.hugegraph.structure.HugeVertex.class));
+ Whitebox.setInternalState(fixture.scheduler, "taskTx", rejecting);
+ Throwable failed = Assert.assertThrows(RuntimeException.class,
+ () -> fixture.scheduler.delete(task.id(), true));
+ Throwable cause = failed;
+ while (cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ Assert.assertSame(rejected, cause);
+ Assert.assertNotNull(fixture.scheduler.findTask(task.id()));
+ Assert.assertNotNull(fixture.scheduler.task(task.id()));
+ Assert.assertEquals(TaskStatus.DELETING, task.status());
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ fixture.scheduler.delete(task.id(), true);
+ assertTaskAbsent(fixture.scheduler, task);
+ job.releaseSave.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ Assert.assertTrue(job.saved.await(10L, TimeUnit.SECONDS));
+ assertTaskAbsent(fixture.scheduler, task);
+ Assert.assertEquals(0, fixture.scheduler.pendingTasks());
+ } finally {
+ job.releaseSave.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ }
+ }
+ }
+
+ @Test
+ public void testUnownedExplicitSaveCanRestoreDeletingRecord() throws Exception {
+ try (Fixture fixture = new Fixture("standard_unowned_explicit_save")) {
+ HugeTask<Object> task = fixture.task(new EmptyJob());
+ task.overwriteStatus(TaskStatus.DELETING);
+ fixture.scheduler.save(task);
+ Assert.assertEquals(0, fixture.scheduler.pendingTasks());
+ Assert.assertEquals(TaskStatus.DELETING, fixture.scheduler.task(task.id()).status());
+ fixture.scheduler.delete(task.id(), true);
+ assertTaskAbsent(fixture.scheduler, task);
+ fixture.scheduler.save(task);
+ Assert.assertEquals(TaskStatus.DELETING, fixture.scheduler.findTask(task.id()).status());
+ Assert.assertEquals(0, fixture.scheduler.pendingTasks());
+ }
+ }
+
+ @Test
+ public void testNonForceDeleteStillRequiresPersistedCompletion() throws Exception {
+ try (Fixture fixture = new Fixture("standard_nonforce_durable_check");
+ SchedulerBinding binding = new SchedulerBinding(fixture)) {
+ CompletionSaveJob job = new CompletionSaveJob();
+ HugeTask<Object> task = fixture.task(job);
+ try {
+ fixture.scheduler.schedule(task);
+ Assert.assertTrue(job.beforeSave.await(10L, TimeUnit.SECONDS));
+ Assert.assertEquals(TaskStatus.SUCCESS, task.status());
+ Assert.assertEquals(TaskStatus.RUNNING, fixture.scheduler.findTask(task.id()).status());
+ Assert.assertThrows(RuntimeException.class, () -> fixture.scheduler.delete(task.id(), false));
+ Assert.assertEquals(TaskStatus.SUCCESS, task.status());
+ Assert.assertEquals(TaskStatus.RUNNING, fixture.scheduler.findTask(task.id()).status());
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ Assert.assertFalse(task.isCancelled());
+ job.releaseSave.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ Assert.assertEquals(TaskStatus.SUCCESS, fixture.scheduler.findTask(task.id()).status());
+ fixture.scheduler.delete(task.id(), false);
+ assertTaskAbsent(fixture.scheduler, task);
+ } finally {
+ job.releaseSave.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ }
+ }
+ }
+
+ @Test
+ public void testNonForceDeleteRejectsActuallyRunningTask() throws Exception {
+ try (Fixture fixture = new Fixture("standard_nonforce_running");
+ SchedulerBinding binding = new SchedulerBinding(fixture)) {
+ BlockingJob job = new BlockingJob();
+ HugeTask<Object> task = fixture.task(job);
+ try {
+ fixture.scheduler.schedule(task);
+ Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS));
+ Assert.assertEquals(TaskStatus.RUNNING, task.status());
+ Assert.assertThrows(IllegalArgumentException.class, () -> fixture.scheduler.delete(task.id(), false));
+ Assert.assertEquals(TaskStatus.RUNNING, task.status());
+ Assert.assertNotNull(fixture.scheduler.findTask(task.id()));
+ Assert.assertEquals(1, fixture.scheduler.pendingTasks());
+ Assert.assertFalse(task.isCancelled());
+ } finally {
+ job.release.countDown();
+ fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS);
+ }
+ }
+ }
+
+ private static void assertTaskAbsent(StandardTaskScheduler scheduler, HugeTask<?> task) {
+ for (boolean withResult : new boolean[]{true, false}) {
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> scheduler.task(task.id(), withResult));
+ Assert.assertFalse(scheduler.tasks(java.util.Collections.singletonList(task.id()), withResult).hasNext());
+ }
+ Assert.assertThrows(org.apache.hugegraph.exception.NotFoundException.class,
+ () -> scheduler.findTask(task.id()));
+ }
+
+ private static final class SchedulerBinding implements AutoCloseable {
+ private final java.util.Map<HugeGraphParams, org.apache.hugegraph.task.TaskScheduler> schedulers;
+ private final HugeGraphParams nativeParams;
+ private final org.apache.hugegraph.task.TaskScheduler original;
+ private final StandardTaskScheduler owner;
+ @SuppressWarnings("unchecked")
+ private SchedulerBinding(Fixture fixture) {
+ this.schedulers = (java.util.Map<HugeGraphParams, org.apache.hugegraph.task.TaskScheduler>)
+ Whitebox.getInternalState(org.apache.hugegraph.task.TaskManager.instance(), "schedulers");
+ this.nativeParams = (HugeGraphParams) Whitebox.getInternalState(fixture.graph, "params");
+ this.owner = fixture.scheduler;
+ this.original = this.schedulers.put(this.nativeParams, this.owner);
+ Assert.assertNotNull(this.original);
+ Assert.assertSame(this.owner, fixture.graph.taskScheduler());
+ }
+ @Override
+ public void close() {
+ Assert.assertTrue(this.schedulers.replace(this.nativeParams, this.owner, this.original));
+ }
+ }
+
private static void await(CountDownLatch latch) {
boolean interrupted = false;
while (true) {
@@ -472,6 +929,34 @@ public class StandardTaskAdmissionTest {
}
}
+ public static class CompletionSaveJob extends EmptyJob {
+ private final CountDownLatch beforeSave = new CountDownLatch(1);
+ private final CountDownLatch releaseSave = new CountDownLatch(1);
+ private final CountDownLatch saved = new CountDownLatch(1);
+ @Override
+ public Object execute() {
+ this.graph().vertices().hasNext();
+ return "completion-result";
+ }
+ @Override
+ protected void done() {
+ this.beforeSave.countDown();
+ await(this.releaseSave);
+ try {
+ super.done();
+ } finally {
+ this.saved.countDown();
+ }
+ }
+ }
+
+ public static class ResultJob extends EmptyJob {
+ @Override
+ public Object execute() {
+ return "ordinary-result";
+ }
+ }
+
public static class BlockingJob extends SysJob<Object> {
private final CountDownLatch started = new CountDownLatch(1);
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerConfigTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerConfigTest.java
index b8601c22d..51628c0d9 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerConfigTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerConfigTest.java
@@ -17,21 +17,3388 @@
package org.apache.hugegraph.unit.core;
+import java.io.File;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.script.ScriptException;
+import javax.script.SimpleBindings;
+
+import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.hugegraph.HugeFactory;
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.StandardHugeGraph;
+import org.apache.hugegraph.auth.AuthContext;
+import org.apache.hugegraph.auth.ContextGremlinServer;
+import org.apache.hugegraph.auth.HugeAuthenticator;
+import org.apache.hugegraph.auth.HugeGraphAuthProxy;
+import org.apache.hugegraph.auth.HugePermission;
+import org.apache.hugegraph.auth.HugeResource;
+import org.apache.hugegraph.auth.RolePermission;
+import org.apache.hugegraph.backend.store.BackendFeatures;
+import org.apache.hugegraph.backend.store.BackendProviderFactory;
+import org.apache.hugegraph.backend.store.BackendStoreProvider;
+import org.apache.hugegraph.backend.store.raft.RaftBackendStoreProvider;
+import org.apache.hugegraph.backend.store.raft.RaftContext;
+import org.apache.hugegraph.backend.store.raft.RaftNode;
+import org.apache.hugegraph.backend.store.rocksdb.RocksDBOptions;
+import org.apache.hugegraph.backend.store.rocksdb.RocksDBStoreProvider;
import org.apache.hugegraph.config.CoreOptions;
import org.apache.hugegraph.config.HugeConfig;
+import org.apache.hugegraph.config.OptionSpace;
+import org.apache.hugegraph.config.RpcOptions;
import org.apache.hugegraph.config.ServerOptions;
import org.apache.hugegraph.core.GraphManager;
import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventListener;
+import org.apache.hugegraph.exception.HugeException;
+import org.apache.hugegraph.rpc.RpcClientProvider;
+import org.apache.hugegraph.rpc.RpcServer;
+import org.apache.hugegraph.rpc.RpcServiceConfig4Client;
+import org.apache.hugegraph.rpc.RpcServiceConfig4Server;
+import org.apache.hugegraph.space.GraphSpace;
+import org.apache.hugegraph.task.TaskManager;
import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.testutil.Whitebox;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Events;
+import org.apache.hugegraph.meta.MetaManager;
+import org.apache.tinkerpop.gremlin.server.Settings;
+import org.apache.tinkerpop.gremlin.server.util.DefaultGraphManager;
+import org.mockito.Mockito;
+import org.apache.tinkerpop.gremlin.groovy.jsr223.GremlinGroovyScriptEngine;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import com.alipay.sofa.jraft.Node;
+import com.alipay.sofa.jraft.NodeManager;
+import com.alipay.sofa.jraft.core.State;
+import com.alipay.sofa.jraft.entity.PeerId;
+import com.alipay.sofa.jraft.option.CliOptions;
+import com.alipay.sofa.jraft.rpc.CliRequests;
+import com.alipay.sofa.jraft.rpc.impl.BoltRpcServer;
+import com.alipay.sofa.jraft.rpc.impl.cli.CliClientServiceImpl;
+import com.alipay.sofa.rpc.config.ServerConfig;
+import com.alipay.sofa.jraft.rpc.RaftRpcServerFactory;
+import com.alipay.sofa.jraft.util.Endpoint;
public class GraphManagerConfigTest {
+ private static ExecutorService previousEventExecutor;
+ private static ExecutorService previousTaskDbExecutor;
+ private static ExecutorService taskDbExecutor;
+
+ @BeforeClass
+ public static void initEventExecutor() {
+ // Other suite classes may have destroyed the global EventHub executor.
+ ExecutorService previous = Whitebox.getInternalState(EventHub.class, "executor");
+ previousEventExecutor = previous != null && !previous.isShutdown() ? previous : null;
+ Whitebox.setInternalState(EventHub.class, "executor", null);
+ EventHub.init(2);
+ previousTaskDbExecutor = Whitebox.getInternalState(TaskManager.instance(), "taskDbExecutor");
+ taskDbExecutor = Executors.newSingleThreadExecutor(
+ runnable -> new Thread(runnable, "task-db-worker-rpc-config-test"));
+ Whitebox.setInternalState(TaskManager.instance(), "taskDbExecutor", taskDbExecutor);
+ }
+
+ @AfterClass
+ public static void restoreEventExecutor() throws InterruptedException {
+ try {
+ Assert.assertTrue(EventHub.destroy(5));
+ } finally {
+ Whitebox.setInternalState(EventHub.class, "executor", previousEventExecutor != null &&
+ !previousEventExecutor.isShutdown() ? previousEventExecutor : null);
+ taskDbExecutor.shutdown();
+ try {
+ Assert.assertTrue(taskDbExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ } finally {
+ Whitebox.setInternalState(TaskManager.instance(), "taskDbExecutor", previousTaskDbExecutor);
+ }
+ }
+ }
+
+ @Test
+ public void testFixturePreservesLivePreviousEventExecutor() throws Exception {
+ this.checkFixtureEventExecutor(false);
+ }
+
+ @Test
+ public void testFixtureDoesNotRestoreDestroyedEventExecutor() throws Exception {
+ this.checkFixtureEventExecutor(true);
+ }
+
+ private void checkFixtureEventExecutor(boolean destroyed) throws Exception {
+ ExecutorService activeEvent = Whitebox.getInternalState(EventHub.class, "executor");
+ ExecutorService activeTask = Whitebox.getInternalState(TaskManager.instance(), "taskDbExecutor");
+ ExecutorService savedPreviousEvent = previousEventExecutor;
+ ExecutorService savedPreviousTask = previousTaskDbExecutor;
+ ExecutorService savedTask = taskDbExecutor;
+ ExecutorService prior = Executors.newSingleThreadExecutor();
+ try {
+ prior.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ if (destroyed) {
+ prior.shutdown();
+ Assert.assertTrue(prior.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ Whitebox.setInternalState(EventHub.class, "executor", prior);
+ initEventExecutor();
+ restoreEventExecutor();
+ EventHub.init(1);
+ EventHub hub = new EventHub("rpc-fixture-executor-control");
+ hub.listen("probe", event -> true);
+ RuntimeException submission = null;
+ try {
+ Assert.assertEquals(Integer.valueOf(1), hub.notify("probe").get(5, TimeUnit.SECONDS));
+ } catch (RuntimeException failure) {
+ submission = failure;
+ }
+ ExecutorService restored = Whitebox.getInternalState(EventHub.class, "executor");
+ Assert.assertNull("Previous destroyed=" + destroyed + ", restoredShutdown=" +
+ restored.isShutdown() + ", notifyFailure=" + submission, submission);
+ if (!destroyed) {
+ Assert.assertSame(prior, restored);
+ } else {
+ Assert.assertNotSame(prior, restored);
+ }
+ } finally {
+ ExecutorService current = Whitebox.getInternalState(EventHub.class, "executor");
+ if (current != null && current != activeEvent) {
+ current.shutdown();
+ Assert.assertTrue(current.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ prior.shutdown();
+ Assert.assertTrue(prior.awaitTermination(5, TimeUnit.SECONDS));
+ Whitebox.setInternalState(EventHub.class, "executor", activeEvent);
+ Whitebox.setInternalState(TaskManager.instance(), "taskDbExecutor", activeTask);
+ previousEventExecutor = savedPreviousEvent;
+ previousTaskDbExecutor = savedPreviousTask;
+ taskDbExecutor = savedTask;
+ }
+ }
+
+ @Rule
+ public TemporaryFolder temporary = new TemporaryFolder();
+
+ @Test
+ public void testRpcFeatureCheckPreservesCallbackTransaction() throws Exception {
+ for (boolean shared : new boolean[]{false, true}) {
+ for (boolean opened : new boolean[]{false, true}) {
+ this.checkRpcCallbackTransaction(shared, opened, false);
+ if (shared) {
+ this.checkRpcCallbackTransaction(true, opened, true);
+ }
+ }
+ }
+ }
+
+ private void checkRpcCallbackTransaction(boolean shared, boolean opened, boolean failing) throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(CoreOptions.BACKEND.name(), "memory");
+ properties.setProperty(CoreOptions.STORE.name(), "callback_tx");
+ StandardHugeGraph graph = shared ? new SharedMemoryGraph(new HugeConfig(properties)) :
+ new StandardHugeGraph(new HugeConfig(properties));
+ RpcServiceConfig4Server server = Mockito.mock(RpcServiceConfig4Server.class);
+ RpcServiceConfig4Client client = Mockito.mock(RpcServiceConfig4Client.class);
+ IllegalStateException failure = new IllegalStateException("RPC proxy failed");
+ if (failing) {
+ Mockito.doThrow(failure).when(client).serviceProxy(Mockito.anyString(), Mockito.any(Class.class));
+ }
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ ExecutorService callback = Executors.newSingleThreadExecutor();
+ try {
+ callback.submit(() -> {
+ try {
+ if (opened) {
+ graph.tx().open();
+ }
+ if (failing) {
+ Assert.assertThrows(IllegalStateException.class,
+ () -> graph.registerRpcServices(server, client),
+ caught -> Assert.assertSame(failure, caught));
+ } else {
+ graph.registerRpcServices(server, client);
+ }
+ Assert.assertEquals(opened, graph.tx().isOpen());
+ AtomicInteger refs = Whitebox.getInternalState(graph.tx(), "refs");
+ Assert.assertEquals(opened ? 1 : 0, refs.get());
+ ThreadLocal<?> transactions = Whitebox.getInternalState(graph.tx(), "transactions");
+ Assert.assertEquals(opened, transactions.get() != null);
+ } finally {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ }
+ }).get(5, TimeUnit.SECONDS);
+ } finally {
+ callback.shutdownNow();
+ Assert.assertTrue(callback.awaitTermination(5, TimeUnit.SECONDS));
+ graph.close();
+ }
+ }
+
+ @Test
+ public void testRpcFeatureFailurePreservesPrimaryException() throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(CoreOptions.BACKEND.name(), "memory");
+ properties.setProperty(CoreOptions.STORE.name(), "feature_failure");
+ FeatureFailureGraph graph = new FeatureFailureGraph(new HugeConfig(properties));
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ IllegalStateException failure = new IllegalStateException("backend feature read failed");
+ IllegalStateException cleanup = new IllegalStateException("transaction listener failed");
+ graph.failure = failure;
+ graph.tx().addTransactionListener(status -> {
+ throw cleanup;
+ });
+ try {
+ Assert.assertThrows(IllegalStateException.class, () -> graph.registerRpcServices(null, null),
+ caught -> Assert.assertSame(failure, caught));
+ Assert.assertFalse(graph.tx().isOpen());
+ Assert.assertEquals(1, failure.getSuppressed().length);
+ Assert.assertSame(cleanup, failure.getSuppressed()[0]);
+ ThreadLocal<?> transactions = Whitebox.getInternalState(graph.tx(), "transactions");
+ Assert.assertNull(transactions.get());
+ } finally {
+ graph.failure = null;
+ graph.tx().clearTransactionListeners();
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ graph.close();
+ }
+ }
+
+ private static class FeatureFailureGraph extends StandardHugeGraph {
+
+ private RuntimeException failure;
+
+ FeatureFailureGraph(HugeConfig config) {
+ super(config);
+ }
+
+ @Override
+ public BackendFeatures backendStoreFeatures() {
+ BackendFeatures features = super.backendStoreFeatures();
+ if (this.failure != null) {
+ throw this.failure;
+ }
+ return features;
+ }
+ }
+
+ @Test
+ public void testLocalLoadRollbackUsesAdminScope() throws Exception {
+ this.checkLocalLoadRollback(false);
+ }
+
+ @Test
+ public void testLocalLoadRollbackRetainsOwnerAfterCloseFailure() throws Exception {
+ this.checkLocalLoadRollback(true);
+ }
+
+ private void checkLocalLoadRollback(boolean foreignTransaction) throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ String name = foreignTransaction ? "rollback_pending" : "rollback_admin";
+ File config = this.temporary.newFile(name + ".properties");
+ Files.write(config.toPath(), ("gremlin.graph=" + TrackedAuthMemoryGraph.class.getName() +
+ "\nbackend=memory\nstore=" + name + "\n").getBytes(StandardCharsets.UTF_8));
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder(name).getPath());
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "127.0.0.1");
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "127.0.0.1:1");
+ GraphManager manager = new GraphManager(new HugeConfig(properties), new EventHub(name));
+ RpcServer server = Mockito.mock(RpcServer.class);
+ Mockito.when(server.enabled()).thenReturn(true);
+ RpcServer originalServer = Whitebox.getInternalState(manager, "rpcServer");
+ Mockito.when(server.config()).thenReturn(originalServer.config());
+ IllegalStateException failure = new IllegalStateException("RPC publication failed");
+ Mockito.doThrow(failure).when(server).exportAll();
+ Whitebox.setInternalState(manager, "rpcServer", server);
+ Whitebox.setInternalState(manager, "rpcInitialized", true);
+ StandardHugeGraph graph = (StandardHugeGraph) HugeFactory.open(new HugeConfig(config.getPath()));
+ graph.initBackend();
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ CountDownLatch ready = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ Future<?> transaction = foreignTransaction ? worker.submit(() -> {
+ graph.tx().open();
+ ready.countDown();
+ try {
+ Assert.assertTrue(release.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(interrupted);
+ } finally {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ }
+ }) : null;
+ String caller = foreignTransaction ? HugeAuthenticator.User.ADMIN.toJson() :
+ new HugeAuthenticator.User("writer", new WriterRole("DEFAULT", name)).toJson();
+ AuthContext.setContext(caller);
+ TaskManager.setContext(caller);
+ HugeGraphAuthProxy.setRequestGraphSpace("DEFAULT");
+ try {
+ if (foreignTransaction) {
+ Assert.assertTrue(ready.await(5, TimeUnit.SECONDS));
+ }
+ manager.loadGraphsFromLocal(Map.of(name, config.getPath()));
+ Assert.assertEquals(caller, AuthContext.getContext());
+ Assert.assertEquals(caller, TaskManager.getContext());
+ Assert.assertEquals("DEFAULT", HugeGraphAuthProxy.getRequestGraphSpace());
+ Map<?, ?> owners = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ if (foreignTransaction) {
+ Assert.assertFalse(graph.closed());
+ Assert.assertSame(graph, owners.get(graph.spaceGraphName()));
+ Assert.assertSame(TrackedAuthMemoryGraph.last, manager.graph(graph.spaceGraphName()));
+ Assert.assertEquals(1, failure.getSuppressed().length);
+ } else {
+ Assert.assertTrue(graph.closed());
+ Assert.assertFalse(owners.containsKey(graph.spaceGraphName()));
+ Assert.assertFalse(manager.graphs().contains(graph.spaceGraphName()));
+ Assert.assertEquals(0, failure.getSuppressed().length);
+ }
+ } finally {
+ release.countDown();
+ if (transaction != null) {
+ transaction.get(5, TimeUnit.SECONDS);
+ }
+ worker.shutdownNow();
+ Assert.assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ HugeGraphAuthProxy.runAsAdmin(() -> {
+ manager.close();
+ try {
+ graph.close();
+ } catch (Exception cleanup) {
+ throw new IllegalStateException(cleanup);
+ }
+ });
+ HugeFactory.remove(graph);
+ TrackedAuthMemoryGraph.last = null;
+ HugeGraphAuthProxy.resetContext();
+ TaskManager.resetContext();
+ }
+ }
+
+ @Test
+ public void testRetainedLoadCandidateIsNotRepublishedAtRpcStartup() throws Exception {
+ this.checkRetainedCandidateRpcStartup(false);
+ }
+
+ @Test
+ public void testRetainedCreateCandidateIsNotRepublishedAtRpcStartup() throws Exception {
+ this.checkRetainedCandidateRpcStartup(true);
+ }
+
+ private void checkRetainedCandidateRpcStartup(boolean create) throws Exception {
+ String name = create ? "failed_create_startup" : "failed_load_startup";
+ GraphManager manager = this.dropManager(name);
+ RpcServer original = Whitebox.getInternalState(manager, "rpcServer");
+ RpcServer server = Mockito.spy(original);
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ Whitebox.setInternalState(manager, "rpcServer", server);
+ Whitebox.setInternalState(manager, "rpcInitialized", true);
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ CountDownLatch ready = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicReference<Future<?>> transaction = new AtomicReference<>();
+ IllegalStateException rejected = new IllegalStateException("candidate RPC export rejected");
+ Mockito.doAnswer(invocation -> {
+ HugeGraph graph = TrackedSharedMemoryGraph.last;
+ Assert.assertNotNull(graph);
+ if (!create) {
+ graph.initBackend();
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ }
+ transaction.set(worker.submit(() -> {
+ graph.tx().open();
+ ready.countDown();
+ try {
+ E.checkState(release.await(10, TimeUnit.SECONDS), "Candidate transaction barrier timed out");
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(interrupted);
+ } finally {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ }
+ }));
+ Assert.assertTrue(ready.await(5, TimeUnit.SECONDS));
+ throw rejected;
+ }).when(server).exportAll();
+ try {
+ String configText = "gremlin.graph=" + TrackedSharedMemoryGraph.class.getName() +
+ "\nbackend=memory\nstore=" + name + "\n";
+ if (create) {
+ Assert.assertThrows(IllegalStateException.class, () -> manager.createGraphLocal(name, configText),
+ caught -> Assert.assertSame(rejected, caught));
+ } else {
+ File config = this.temporary.newFile(name + ".properties");
+ Files.write(config.toPath(), configText.getBytes(StandardCharsets.UTF_8));
+ manager.loadGraphsFromLocal(Map.of(name, config.getPath()));
+ }
+ HugeGraph candidate = TrackedSharedMemoryGraph.last;
+ Assert.assertFalse(candidate.closed());
+ Assert.assertSame(candidate, manager.graph(candidate.spaceGraphName()));
+ Assert.assertEquals(1, rejected.getSuppressed().length);
+ assertRpcResources(server, client, 0);
+ Mockito.doCallRealMethod().when(server).exportAll();
+ Whitebox.setInternalState(manager, "rpcInitialized", false);
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ Assert.assertSame(candidate, manager.graph(candidate.spaceGraphName()));
+ Assert.assertFalse(candidate.closed());
+ assertRpcResources(server, client, 0);
+ } finally {
+ release.countDown();
+ if (transaction.get() != null) {
+ transaction.get().get(5, TimeUnit.SECONDS);
+ }
+ worker.shutdown();
+ Assert.assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ manager.close();
+ if (TrackedSharedMemoryGraph.last != null) {
+ TrackedSharedMemoryGraph.last.close();
+ HugeFactory.remove(TrackedSharedMemoryGraph.last);
+ }
+ TrackedSharedMemoryGraph.last = null;
+ }
+ }
+
+ public static class TrackedSharedMemoryGraph extends SharedMemoryGraph {
+
+ private static TrackedSharedMemoryGraph last;
+
+ public TrackedSharedMemoryGraph(HugeConfig config) {
+ super(config);
+ }
+
+ public static TrackedSharedMemoryGraph open(Configuration config) {
+ last = new TrackedSharedMemoryGraph(new HugeConfig(config));
+ return last;
+ }
+ }
+
+ public static class TrackedAuthMemoryGraph {
+
+ private static HugeGraph last;
+
+ public static HugeGraph open(Configuration config) {
+ last = new HugeGraphAuthProxy(HugeFactory.open(config));
+ return last;
+ }
+ }
+
+ @Test
+ public void testAuthenticatedRpcCleanupPreservesOwnerUntilAdmin() throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder("auth").getPath());
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false);
+ properties.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "127.0.0.1");
+ properties.setProperty(RpcOptions.RPC_SERVER_PORT.name(), 0);
+ try (ServerSocket reservation = new ServerSocket(0)) {
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "127.0.0.1:" + reservation.getLocalPort());
+ }
+ EventHub hub = new EventHub("authenticated-rpc-cleanup");
+ GraphManager manager = new GraphManager(new HugeConfig(properties), hub);
+ String originalThreadName = Thread.currentThread().getName();
+ AuthContext.setContext(HugeAuthenticator.User.ADMIN.toJson());
+ TaskManager.setContext(HugeAuthenticator.User.ADMIN.toJson());
+ try {
+ String factory = "gremlin.graph=" + AuthSharedMemoryGraph.class.getName() + "\nbackend=memory\nstore=";
+ manager.createGraphLocal("auth_startup", factory + "auth_startup\n");
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ HugeGraph owner = manager.createGraphLocal("auth_late", factory + "auth_late\n");
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 4);
+ EventListener rejection = event -> {
+ throw new IllegalStateException("reject authenticated candidate");
+ };
+ hub.listen(Events.GRAPH_CREATE, rejection);
+ try {
+ Assert.assertThrows(HugeException.class, () -> manager.createGraphLocal(
+ "auth_rejected", factory + "auth_rejected\n"));
+ } finally {
+ hub.unlisten(Events.GRAPH_CREATE, rejection);
+ }
+ assertRpcResources(server, client, 4);
+ TaskManager.resetContext();
+ RolePermission execute = RolePermission.role(owner.graphSpace(), owner.name(), HugePermission.EXECUTE);
+ Assert.assertFalse(RolePermission.isAdmin(execute));
+ AuthContext.setContext(new HugeAuthenticator.User("reader", execute).toJson());
+ SimpleBindings bindings = new SimpleBindings();
+ bindings.put("graph", owner);
+ GremlinGroovyScriptEngine engine = new GremlinGroovyScriptEngine();
+ Assert.assertThrows(ScriptException.class, () -> engine.eval("graph.unregisterRpcServices()", bindings),
+ error -> Assert.assertContains("Permission denied", error.getMessage()));
+ assertRpcResources(server, client, 4);
+ AuthContext.resetContext();
+ Assert.assertThrows(IllegalStateException.class, owner::unregisterRpcServices,
+ error -> Assert.assertContains("Missing authentication context", error.getMessage()));
+ assertRpcResources(server, client, 4);
+ HugeGraphAuthProxy.runAsAdmin(owner::unregisterRpcServices);
+ assertRpcResources(server, client, 2);
+ AuthContext.setContext(HugeAuthenticator.User.ADMIN.toJson());
+ // Use the normal PD-drop branch with explicit admin identity and no placement metadata writes.
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ spaces.put("DEFAULT", new GraphSpace("DEFAULT"));
+ Thread.currentThread().setName("grpc-auth-cleanup-test");
+ manager.dropGraph("DEFAULT", "auth_late", false);
+ Assert.assertTrue(owner.closed());
+ assertRpcResources(server, client, 2);
+ } finally {
+ Thread.currentThread().setName(originalThreadName);
+ try {
+ HugeGraphAuthProxy.runAsAdmin(manager::close);
+ } finally {
+ HugeGraphAuthProxy.resetContext();
+ TaskManager.resetContext();
+ }
+ }
+ }
+
+ @Test
+ public void testAuthenticatedPublicationPreservesCallerScope() throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder("lazy").getPath());
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "127.0.0.1");
+ properties.setProperty(RpcOptions.RPC_SERVER_PORT.name(), 0);
+ try (ServerSocket reservation = new ServerSocket(0)) {
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "127.0.0.1:" + reservation.getLocalPort());
+ }
+ GraphManager manager = new GraphManager(new HugeConfig(properties), new EventHub("lazy-auth-rpc"));
+ SharedMemoryGraph warm = null;
+ try {
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ String space = "rpcscope";
+ String name = "lazy_auth";
+ PropertiesConfiguration config = new PropertiesConfiguration();
+ config.setProperty(CoreOptions.BACKEND.name(), "memory");
+ config.setProperty(CoreOptions.STORE.name(), name);
+ config.setProperty(CoreOptions.GRAPH_SPACE.name(), space);
+ warm = new SharedMemoryGraph(new HugeConfig(config));
+ warm.initBackend();
+ HugeGraph owner = new HugeGraphAuthProxy(warm);
+ RolePermission role = new WriterRole(space, name);
+ Assert.assertFalse(RolePermission.isAdmin(role));
+ String caller = new HugeAuthenticator.User("writer", role).toJson();
+ AuthContext.setContext(caller);
+ TaskManager.setContext(caller);
+ HugeGraphAuthProxy.setRequestGraphSpace(space);
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{String.class, HugeGraph.class},
+ "publishGraph", manager, owner.spaceGraphName(), owner);
+ Assert.assertSame(owner, manager.graph(space + "-" + name));
+ Assert.assertEquals(caller, AuthContext.getContext());
+ Assert.assertEquals(caller, TaskManager.getContext());
+ Assert.assertEquals(space, HugeGraphAuthProxy.getRequestGraphSpace());
+ Assert.assertEquals("writer", HugeGraphAuthProxy.username());
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 2);
+ Assert.assertThrows(jakarta.ws.rs.ForbiddenException.class, owner::unregisterRpcServices);
+ assertRpcResources(server, client, 2);
+ } finally {
+ HugeGraph retained = warm;
+ HugeGraphAuthProxy.runAsAdmin(() -> {
+ manager.close();
+ if (retained != null) {
+ try {
+ retained.close();
+ } catch (Exception failure) {
+ throw new IllegalStateException(failure);
+ } finally {
+ HugeFactory.remove(retained);
+ }
+ }
+ });
+ HugeGraphAuthProxy.resetContext();
+ TaskManager.resetContext();
+ }
+ }
+
+ private static class WriterRole extends RolePermission {
+
+ WriterRole(String space, String graph) {
+ this.add(space, graph, HugePermission.READ, HugeResource.ALL_RES);
+ this.add(space, graph, HugePermission.WRITE, HugeResource.ALL_RES);
+ }
+ }
+
+ public static class AuthSharedMemoryGraph {
+
+ public static HugeGraph open(Configuration config) {
+ return new HugeGraphAuthProxy(new SharedMemoryGraph(new HugeConfig(config)));
+ }
+ }
+
+ @Test
+ public void testLegacyDefaultCleanupDoesNotAbortDropWithoutRpc() throws Exception {
+ this.checkLegacyDrop(false, false);
+ }
+
+ @Test
+ public void testActualCleanupFailureStillPropagates() throws Exception {
+ this.checkLegacyDrop(true, false);
+ }
+
+ @Test
+ public void testClearCleanupFailurePreservesOwnerForRetry() throws Exception {
+ this.checkLegacyDrop(true, true);
+ }
+
+ @Test
+ public void testQuotaUpdateFailurePreservesOwnerAndCountForRetry() throws Exception {
+ this.checkLegacyDrop(false, false, true);
+ }
+
+ private void checkLegacyDrop(boolean failCleanup, boolean clear) throws Exception {
+ this.checkLegacyDrop(failCleanup, clear, false);
+ }
+
+ private void checkLegacyDrop(boolean failCleanup, boolean clear, boolean failQuota) throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder("legacy").getPath());
+ properties.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false);
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "");
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "");
+ EventHub hub = new EventHub("legacy-rpc-cleanup-test");
+ GraphManager manager = new GraphManager(new HugeConfig(properties), hub);
+ HugeGraph delegate = null;
+ String originalThreadName = Thread.currentThread().getName();
+ try {
+ delegate = manager.createGraphLocal("legacy_cleanup", "gremlin.graph=" + HugeFactory.class.getName() +
+ "\nbackend=memory\nstore=legacy_cleanup\n");
+ HugeGraph owner = delegate;
+ IllegalStateException failure = new IllegalStateException("actual owner cleanup failed");
+ AtomicBoolean failing = new AtomicBoolean(failCleanup);
+ HugeGraph legacy = (HugeGraph) Proxy.newProxyInstance(HugeGraph.class.getClassLoader(),
+ new Class<?>[]{HugeGraph.class}, (proxy, method, args) -> {
+ if ("unregisterRpcServices".equals(method.getName())) {
+ if (failing.get()) {
+ throw failure;
+ }
+ return InvocationHandler.invokeDefault(proxy, method, args);
+ }
+ try {
+ return method.invoke(owner, args);
+ } catch (InvocationTargetException error) {
+ throw error.getCause();
+ }
+ });
+ Map<String, HugeGraph> graphs = Whitebox.getInternalState(manager, "graphs");
+ graphs.put(legacy.spaceGraphName(), legacy);
+ DefaultGraphManager bindings = new DefaultGraphManager(new Settings());
+ bindings.putGraph(legacy.spaceGraphName(), legacy);
+ CountDownLatch dropped = new CountDownLatch(1);
+ hub.listen(Events.GRAPH_DROP, event -> {
+ if (bindings.getGraph(legacy.spaceGraphName()) == legacy) {
+ try {
+ bindings.removeGraph(legacy.spaceGraphName());
+ } catch (Exception failureToUnbind) {
+ throw new IllegalStateException(failureToUnbind);
+ }
+ }
+ dropped.countDown();
+ return true;
+ });
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ Assert.assertFalse(server.enabled());
+ Assert.assertFalse(client.enabled());
+ // Reach the normal PD-drop branch without booting PD or writing placement metadata.
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ spaces.put("DEFAULT", space);
+ if (failQuota) {
+ Mockito.doThrow(failure).doNothing().when(meta).updateGraphSpaceConfig("DEFAULT", space);
+ } else {
+ Thread.currentThread().setName("grpc-legacy-cleanup-test");
+ }
+ if (failCleanup || failQuota) {
+ Assert.assertThrows(IllegalStateException.class, () -> manager.dropGraph(
+ "DEFAULT", "legacy_cleanup", clear), caught -> Assert.assertSame(failure, caught));
+ Assert.assertEquals(!clear || failQuota, owner.closed());
+ Assert.assertSame(legacy, manager.graph(legacy.spaceGraphName()));
+ Assert.assertSame(legacy, bindings.getGraph(legacy.spaceGraphName()));
+ Map<?, ?> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertSame(owner, factory.get(legacy.spaceGraphName()));
+ Assert.assertEquals(1L, dropped.getCount());
+ if (failCleanup) {
+ Mockito.verifyNoInteractions(meta);
+ }
+ Assert.assertEquals(5, space.graphNumberUsed());
+ failing.set(false);
+ }
+ manager.dropGraph("DEFAULT", "legacy_cleanup", clear);
+ Assert.assertTrue(dropped.await(5, TimeUnit.SECONDS));
+ Assert.assertTrue(owner.closed());
+ Map<?, ?> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertFalse(factory.containsKey(legacy.spaceGraphName()));
+ Assert.assertNull(bindings.getGraph(legacy.spaceGraphName()));
+ Assert.assertFalse(manager.graphs().contains(legacy.spaceGraphName()));
+ if (failQuota) {
+ Assert.assertEquals(4, space.graphNumberUsed());
+ Mockito.verify(meta, Mockito.times(2)).updateGraphSpaceConfig("DEFAULT", space);
+ }
+ } finally {
+ Thread.currentThread().setName(originalThreadName);
+ manager.close();
+ if (delegate != null) {
+ delegate.close();
+ HugeFactory.remove(delegate);
+ }
+ }
+ }
+
+ @Test
+ public void testPeerRemovalBatchContinuesAfterOpenTransaction() throws Exception {
+ GraphManager manager = this.dropManager("peer_batch");
+ HugeGraph busy = null;
+ HugeGraph next = null;
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ CountDownLatch opened = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ String threadName = Thread.currentThread().getName();
+ try {
+ busy = manager.createGraphLocal("peer_busy", sharedMemoryConfig("peer_busy"));
+ next = manager.createGraphLocal("peer_next", sharedMemoryConfig("peer_next"));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Mockito.when(meta.extractGraphsFromResponse("batch")).thenReturn(
+ List.of(busy.spaceGraphName(), next.spaceGraphName()));
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Thread.currentThread().setName("grpc-peer-removal-test");
+ HugeGraph owner = busy;
+ Future<?> transaction = worker.submit(() -> {
+ try {
+ owner.tx().open();
+ opened.countDown();
+ E.checkState(release.await(10, TimeUnit.SECONDS), "Peer transaction barrier timed out");
+ } catch (InterruptedException failure) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(failure);
+ } finally {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", owner);
+ }
+ });
+ Assert.assertTrue(opened.await(5, TimeUnit.SECONDS));
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphRemoveHandler", manager, "batch");
+ HugeException refusal = Assert.assertThrows(HugeException.class, () -> manager.dropGraph(
+ "DEFAULT", "peer_busy", false));
+ Assert.assertTrue(refusal.getCause() instanceof IllegalStateException);
+ Assert.assertContains("Ensure scheduler and tx closed", refusal.getCause().getMessage());
+ Assert.assertFalse(busy.closed());
+ Assert.assertSame(busy, manager.graph(busy.spaceGraphName()));
+ Assert.assertTrue(next.closed());
+ Assert.assertFalse(manager.graphs().contains(next.spaceGraphName()));
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 2);
+ release.countDown();
+ transaction.get(5, TimeUnit.SECONDS);
+ } finally {
+ release.countDown();
+ worker.shutdownNow();
+ Assert.assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ Thread.currentThread().setName(threadName);
+ manager.close();
+ if (busy != null) {
+ HugeFactory.remove(busy);
+ }
+ if (next != null) {
+ HugeFactory.remove(next);
+ }
+ }
+ }
+
+ @Test
+ public void testClosedGraphCloseFailureRetainsOwnerUntilRetry() throws Exception {
+ this.checkPostCloseDrop(false);
+ }
+
+ @Test
+ public void testClosedGraphRpcFailureRetriesWithClear() throws Exception {
+ this.checkPostCloseDrop(true);
+ }
+
+ @Test
+ public void testReceivedSameNameAddLoadsLatestAfterOwnerCleanup() throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(true);
+ }
+
+ @Test
+ public void testSameNameAddAfterOwnerCleanupLoadsLatestConfig() throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(false);
+ }
+
+ @Test
+ public void testReceivedSameNameAddDoesNotReloadMissingConfig() throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(true, 1);
+ }
+
+ @Test
+ public void testReplacementLoadFailureDoesNotFailCompletedDelete() throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(true, 2);
+ }
+
+ @Test
+ public void testPeerRemoveFailureRetriesOnSameNameAdd() throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(true, 0, true);
+ }
+
+ @Test
+ public void testLateSelfAddRetriesFailedPeerRemoval() throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(false, 0, true, true);
+ }
+
+ @Test
+ public void testPeerRemoveFailureDoesNotReloadMissingConfig() throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(true, 1, true);
+ }
+
+ private void checkReceivedSameNameAddAfterCleanup(boolean addDuringCleanup) throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(addDuringCleanup, 0);
+ }
+
+ private void checkReceivedSameNameAddAfterCleanup(boolean addDuringCleanup, int replayFailure) throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(addDuringCleanup, replayFailure, false);
+ }
+
+ private void checkReceivedSameNameAddAfterCleanup(boolean addDuringCleanup, int replayFailure,
+ boolean peer) throws Exception {
+ this.checkReceivedSameNameAddAfterCleanup(addDuringCleanup, replayFailure, peer, false);
+ }
+
+ private void checkReceivedSameNameAddAfterCleanup(boolean addDuringCleanup, int replayFailure,
+ boolean peer, boolean lateSelfAdd) throws Exception {
+ String name = lateSelfAdd ? "late_self_add_peer_retry" :
+ replayFailure == 1 ? "received_add_missing" : replayFailure == 2 ? "received_add_retry" :
+ addDuringCleanup ? "received_add_early" : "received_add_late";
+ String key = "DEFAULT-" + name;
+ GraphManager manager = this.dropManager(name);
+ EventHub hub = Whitebox.getInternalState(manager, "eventHub");
+ Settings settings = new Settings();
+ settings.gremlinPool = 1;
+ settings.threadPoolBoss = 1;
+ settings.threadPoolWorker = 1;
+ ContextGremlinServer gremlin = new ContextGremlinServer(settings, hub);
+ org.apache.tinkerpop.gremlin.server.GraphManager bindings =
+ gremlin.getServerGremlinExecutor().getGraphManager();
+ HugeGraph owner = null;
+ HugeGraph replacement = null;
+ AtomicBoolean reject = new AtomicBoolean(true);
+ AtomicBoolean rejectReplay = new AtomicBoolean();
+ IllegalStateException readFailure = new IllegalStateException("replacement metadata read rejected");
+ AtomicInteger cleanupAttempts = new AtomicInteger();
+ AtomicInteger singleConfigReads = new AtomicInteger();
+ AtomicInteger allConfigReads = new AtomicInteger();
+ AtomicReference<Map<String, Object>> latest = new AtomicReference<>();
+ try {
+ owner = manager.createGraphLocal(name, "gremlin.graph=" + HugeFactory.class.getName() +
+ "\nbackend=memory\nstore=" + name + "\nalias.graph.id=old-config-marker\n");
+ HugeGraph graph = owner;
+ Map<String, HugeGraph> factories = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertSame(graph, factories.get(key));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ // Memory is not a shared backend: actual cache-RPC configuration stays empty.
+ assertRpcResources(server, client, 0);
+ BackendStoreProvider provider = Mockito.spy(owner.storeProvider());
+ Whitebox.setInternalState(owner, "storeProvider", provider);
+ IllegalStateException failure = new IllegalStateException("old owner cleanup is retained");
+ Mockito.doAnswer(invocation -> {
+ invocation.callRealMethod();
+ cleanupAttempts.incrementAndGet();
+ if (reject.get()) {
+ throw failure;
+ }
+ return null;
+ }).when(provider).close();
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ Mockito.when(meta.graphSpace("DEFAULT")).thenReturn(space);
+ Mockito.when(meta.extractGraphsFromResponse("new-configuration")).thenReturn(
+ java.util.Collections.singletonList(key));
+ Mockito.when(meta.extractGraphsFromResponse("remove-configuration")).thenReturn(
+ java.util.Collections.singletonList(key));
+ Mockito.when(meta.getGraphConfig("DEFAULT", name)).thenAnswer(invocation -> {
+ singleConfigReads.incrementAndGet();
+ if (rejectReplay.get()) {
+ throw readFailure;
+ }
+ Map<String, Object> current = latest.get();
+ return current == null ? null : new HashMap<>(current);
+ });
+ Mockito.when(meta.graphConfigs("DEFAULT")).thenAnswer(invocation -> {
+ allConfigReads.incrementAndGet();
+ Map<String, Object> current = latest.get();
+ Map<String, Map<String, Object>> configs = new HashMap<>();
+ if (current != null) {
+ configs.put(key, new HashMap<>(current));
+ }
+ return configs;
+ });
+ Mockito.doAnswer(invocation -> {
+ latest.set(null);
+ return null;
+ }).when(meta).removeGraphConfig("DEFAULT", name);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Whitebox.setInternalState(manager, "pdPeers", "127.0.0.1:8686");
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ spaces.put("DEFAULT", space);
+ if (lateSelfAdd) {
+ // Local creation is still waiting for its own PD acknowledgement.
+ java.util.Set<String> creating = Whitebox.getInternalState(manager, "creatingGraphs");
+ creating.add(key);
+ }
+ String threadName = Thread.currentThread().getName();
+ if (peer) {
+ try {
+ Thread.currentThread().setName("grpc-peer-removal-control");
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphRemoveHandler", manager, "remove-configuration");
+ } finally {
+ Thread.currentThread().setName(threadName);
+ }
+ Assert.assertThrows(HugeException.class, () -> new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph"));
+ } else {
+ Assert.assertThrows(HugeException.class, () -> new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph"),
+ caught -> Assert.assertSame(failure, caught.getCause()));
+ }
+ Assert.assertTrue(graph.closed());
+ Assert.assertSame(graph, manager.graph(key));
+ Assert.assertSame(graph, factories.get(key));
+ Assert.assertNull(latest.get());
+ Map<String, Object> newConfig = new HashMap<>();
+ newConfig.put(CoreOptions.BACKEND.name(), "memory");
+ newConfig.put(CoreOptions.STORE.name(), name);
+ // This is a latest-config marker; Memory does not implement a physical HStore GraphID.
+ newConfig.put(CoreOptions.ALIAS_NAME.name(), "new-config-marker");
+ newConfig.put("nickname", "new_generation");
+ latest.set(newConfig);
+ if (addDuringCleanup) {
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "new-configuration");
+ Assert.assertSame(graph, manager.graph(key));
+ Assert.assertEquals(0, singleConfigReads.get());
+ }
+ if (replayFailure == 1) {
+ latest.set(null);
+ } else if (replayFailure == 2) {
+ rejectReplay.set(true);
+ }
+ reject.set(false);
+ if (lateSelfAdd) {
+ java.util.Set<HugeGraph> failed = Whitebox.getInternalState(manager, "failedPeerRemovals");
+ Assert.assertTrue(failed.contains(graph));
+ }
+ // A replacement read failure must not turn completed old-owner deletion into a failed DELETE.
+ if (peer) {
+ try {
+ Thread.currentThread().setName("grpc-peer-add-control");
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "new-configuration");
+ } finally {
+ Thread.currentThread().setName(threadName);
+ }
+ Mockito.verify(meta, Mockito.never()).removeGraphConfig("DEFAULT", name);
+ Mockito.verify(meta, Mockito.never()).notifyGraphRemove("DEFAULT", name);
+ Assert.assertEquals(5, space.graphNumberUsed());
+ } else {
+ new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph");
+ }
+ Assert.assertEquals(peer && addDuringCleanup ? 3 : 2, cleanupAttempts.get());
+ Assert.assertNotSame(graph, factories.get(key));
+ java.util.Set<String> pending = Whitebox.getInternalState(manager, "pendingGraphAdds");
+ if (replayFailure == 1) {
+ Assert.assertNull(manager.graph(key));
+ Assert.assertNull(manager.graph("DEFAULT", name));
+ Assert.assertFalse(factories.containsKey(key));
+ Assert.assertFalse(pending.contains(key));
+ return;
+ }
+ Assert.assertSame(newConfig, latest.get());
+ if (replayFailure == 2) {
+ Assert.assertNull(manager.graph(key));
+ Assert.assertFalse(factories.containsKey(key));
+ Assert.assertTrue(pending.contains(key));
+ Assert.assertEquals(1, singleConfigReads.get());
+ rejectReplay.set(false);
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "new-configuration");
+ }
+ if (!addDuringCleanup && !peer) {
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "new-configuration");
+ }
+ HugeGraph automatic = manager.graph(key);
+ boolean listedBeforePublicRead = manager.graphs().contains(key);
+ int configReadsBeforePublicRead = singleConfigReads.get();
+ // The public two-argument lookup has an existing lazy metadata loading fallback.
+ replacement = manager.graph("DEFAULT", name);
+ Assert.assertNotNull(replacement);
+ Assert.assertNotSame(graph, replacement);
+ Assert.assertSame(replacement, manager.graph(key));
+ Assert.assertSame(replacement, factories.get(key));
+ Assert.assertEquals("new-config-marker", replacement.configuration().get(CoreOptions.ALIAS_NAME));
+ Assert.assertEquals("new_generation", replacement.nickname());
+ Assert.assertEquals(0L, replacement.traversal().V().count().next().longValue());
+ Assert.assertTrue(replacement.started());
+ Assert.assertFalse(pending.contains(key));
+ Assert.assertSame(replacement, bindings.getGraph(key));
+ Assert.assertNotNull(bindings.getTraversalSource("__g_" + key));
+ Assert.assertSame(replacement, bindings.getAsBindings().get(key));
+ replacement.schema().vertexLabel("replacement_marker").useAutomaticId().create();
+ replacement.addVertex(org.apache.tinkerpop.gremlin.structure.T.label, "replacement_marker");
+ replacement.tx().commit();
+ // The old asynchronous DROP may arrive after the new CREATE; native bindings check identity.
+ hub.notifySync(Events.GRAPH_DROP, graph);
+ Assert.assertSame(replacement, bindings.getGraph(key));
+ Assert.assertNotNull(bindings.getTraversalSource("__g_" + key));
+ Assert.assertSame(replacement, bindings.getAsBindings().get(key));
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "new-configuration");
+ Assert.assertSame(replacement, manager.graph(key));
+ Assert.assertEquals(1L, replacement.traversal().V().count().next().longValue());
+ assertRpcResources(server, client, 0);
+ String facts = String.format("same-name add during cleanup=%s automaticOwner=%s listedBeforeRead=%s " +
+ "getConfigBeforeRead=%s graphConfigsReads=%s publicLazyLoadedLatest=%s replacementStarted=%s " +
+ "rpcProviderConfigs=%s", addDuringCleanup, automatic == null ? "absent" : "new",
+ listedBeforePublicRead, configReadsBeforePublicRead, allConfigReads.get(),
+ replacement != graph, replacement.started(), server.config().configs().size());
+ Assert.assertNotNull(facts, automatic);
+ Assert.assertTrue(facts, listedBeforePublicRead);
+ } finally {
+ reject.set(false);
+ rejectReplay.set(false);
+ manager.close();
+ gremlin.stop().get(30, TimeUnit.SECONDS);
+ if (owner != null) {
+ owner.close();
+ HugeFactory.remove(owner);
+ }
+ if (replacement != null) {
+ replacement.close();
+ HugeFactory.remove(replacement);
+ }
+ }
+ }
+
+ @Test
+ public void testReloadAfterRestCreateWithoutSelfAdd() throws Exception {
+ this.checkReloadAfterRestCreate(false, false);
+ }
+
+ @Test
+ public void testReloadAfterRestCreateAndSelfAdd() throws Exception {
+ this.checkReloadAfterRestCreate(true, false);
+ }
+
+ @Test
+ public void testReloadAfterHealthyDuplicateAdds() throws Exception {
+ this.checkReloadAfterRestCreate(true, true);
+ }
+
+ private void checkReloadAfterRestCreate(boolean selfAdd, boolean duplicateAdd) throws Exception {
+ String name = duplicateAdd ? "duplicate_add_reload" : selfAdd ? "self_add_reload" : "reload_without_add";
+ String other = name + "_other";
+ String key = "DEFAULT-" + name;
+ GraphManager manager = this.dropManager(name);
+ EventHub hub = Whitebox.getInternalState(manager, "eventHub");
+ Settings settings = new Settings();
+ settings.gremlinPool = 1;
+ settings.threadPoolBoss = 1;
+ settings.threadPoolWorker = 1;
+ ContextGremlinServer gremlin = new ContextGremlinServer(settings, hub);
+ org.apache.tinkerpop.gremlin.server.GraphManager bindings =
+ gremlin.getServerGremlinExecutor().getGraphManager();
+ Map<String, Map<String, Object>> configs = new HashMap<>();
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ GraphSpace space = new GraphSpace("DEFAULT");
+ Mockito.when(meta.graphSpace("DEFAULT")).thenReturn(space);
+ Mockito.when(meta.extractGraphsFromResponse("self-add")).thenReturn(List.of(key, "DEFAULT-" + other));
+ Mockito.when(meta.extractGraphsFromResponse("foreign-add")).thenReturn(List.of(key, "DEFAULT-" + other));
+ Mockito.when(meta.graphConfigs("DEFAULT")).thenAnswer(invocation -> new HashMap<>(configs));
+ Mockito.when(meta.getGraphConfig(Mockito.eq("DEFAULT"), Mockito.anyString()))
+ .thenAnswer(invocation -> {
+ Map<String, Object> config = configs.get("DEFAULT-" + invocation.getArgument(1));
+ return config == null ? null : new HashMap<>(config);
+ });
+ Mockito.doAnswer(invocation -> {
+ Map<String, Object> config = invocation.getArgument(2);
+ configs.put("DEFAULT-" + invocation.getArgument(1), new HashMap<>(config));
+ return null;
+ }).when(meta).addGraphConfig(Mockito.eq("DEFAULT"), Mockito.anyString(), Mockito.anyMap());
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Whitebox.setInternalState(manager, "pdPeers", "127.0.0.1:8686");
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ spaces.put("DEFAULT", space);
+ try {
+ org.apache.hugegraph.api.profile.GraphsAPI api = new org.apache.hugegraph.api.profile.GraphsAPI();
+ for (String graphName : List.of(name, other)) {
+ Map<String, Object> config = new HashMap<>();
+ config.put(CoreOptions.BACKEND.name(), "memory");
+ config.put(CoreOptions.SERIALIZER.name(), "text");
+ api.create(manager, "DEFAULT", graphName, null, config);
+ }
+ HugeGraph owner = manager.graph(key);
+ HugeGraph otherOwner = manager.graph("DEFAULT-" + other);
+ Assert.assertSame(owner, bindings.getGraph(key));
+ if (selfAdd) {
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "self-add");
+ }
+ java.util.Set<String> pending = Whitebox.getInternalState(manager, "pendingGraphAdds");
+ if (duplicateAdd) {
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "foreign-add");
+ Assert.assertSame(owner, manager.graph(key));
+ Assert.assertSame(otherOwner, manager.graph("DEFAULT-" + other));
+ Assert.assertTrue(pending.contains(key));
+ Assert.assertTrue(pending.contains("DEFAULT-" + other));
+ }
+ Mockito.clearInvocations(meta);
+ long begin = System.nanoTime();
+ System.out.println(String.format("self-add reload before: selfAdd=%s pending=%s graphs=%s",
+ selfAdd, pending.contains(key), manager.graphs().size()));
+ try {
+ api.reload(manager, Map.of("action", "reload"));
+ } finally {
+ System.out.println(String.format("self-add reload after: selfAdd=%s durationMs=%s pending=%s graphs=%s",
+ selfAdd, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - begin),
+ pending.contains(key), manager.graphs().size()));
+ }
+ Assert.assertTrue(owner.closed());
+ Assert.assertTrue(otherOwner.closed());
+ Assert.assertNotNull(manager.graph(key));
+ Assert.assertNotNull(manager.graph("DEFAULT-" + other));
+ Assert.assertNotSame(owner, manager.graph(key));
+ Assert.assertNotSame(otherOwner, manager.graph("DEFAULT-" + other));
+ Assert.assertEquals(2, manager.graphs().size());
+ Assert.assertFalse(pending.contains(key));
+ Assert.assertFalse(pending.contains("DEFAULT-" + other));
+ Mockito.verify(meta, Mockito.never()).getGraphConfig(Mockito.anyString(), Mockito.anyString());
+ Mockito.verify(meta).graphConfigs("DEFAULT");
+ Assert.assertSame(manager.graph(key), bindings.getGraph(key));
+ Assert.assertSame(manager.graph("DEFAULT-" + other), bindings.getGraph("DEFAULT-" + other));
+ Mockito.verify(meta, Mockito.never()).removeGraphConfig(Mockito.anyString(), Mockito.anyString());
+ } finally {
+ List<HugeGraph> owners = new ArrayList<>();
+ for (String graphName : manager.graphs()) {
+ owners.add(manager.graph(graphName));
+ }
+ manager.close();
+ gremlin.stop().get(30, TimeUnit.SECONDS);
+ for (HugeGraph owner : owners) {
+ HugeFactory.remove(owner);
+ }
+ }
+ }
+
+ @Test
+ public void testHealthyOwnerAddBeforePeerRemoveLoadsLatestConfig() throws Exception {
+ this.checkAddBeforePeerRemove(false, true);
+ }
+
+ @Test
+ public void testHealthyOwnerAddBeforePeerRemoveDoesNotReviveMissingConfig() throws Exception {
+ this.checkAddBeforePeerRemove(true, true);
+ }
+
+ @Test
+ public void testDuplicateAddWithoutRemoveKeepsHealthyOwner() throws Exception {
+ this.checkAddBeforePeerRemove(false, false);
+ }
+
+ private void checkAddBeforePeerRemove(boolean missingConfig, boolean removeReceived) throws Exception {
+ String name = missingConfig ? "add_before_remove_missing" :
+ removeReceived ? "add_before_remove" : "healthy_duplicate_add";
+ String key = "DEFAULT-" + name;
+ GraphManager manager = this.dropManager(name);
+ EventHub hub = Whitebox.getInternalState(manager, "eventHub");
+ Settings settings = new Settings();
+ settings.gremlinPool = 1;
+ settings.threadPoolBoss = 1;
+ settings.threadPoolWorker = 1;
+ ContextGremlinServer gremlin = new ContextGremlinServer(settings, hub);
+ org.apache.tinkerpop.gremlin.server.GraphManager bindings =
+ gremlin.getServerGremlinExecutor().getGraphManager();
+ HugeGraph owner = null;
+ HugeGraph replacement = null;
+ try {
+ owner = manager.createGraphLocal(name, "gremlin.graph=" + HugeFactory.class.getName() +
+ "\nbackend=memory\nstore=" + name + "\nalias.graph.id=old-before-remove\n");
+ Map<String, HugeGraph> factories = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ owner.schema().vertexLabel("old_marker").useAutomaticId().create();
+ owner.addVertex(org.apache.tinkerpop.gremlin.structure.T.label, "old_marker");
+ owner.tx().commit();
+ Map<String, Object> latest = new HashMap<>();
+ latest.put(CoreOptions.BACKEND.name(), "memory");
+ latest.put(CoreOptions.STORE.name(), name);
+ latest.put(CoreOptions.ALIAS_NAME.name(), "new-before-remove");
+ AtomicInteger configReads = new AtomicInteger();
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ Mockito.when(meta.graphSpace("DEFAULT")).thenReturn(space);
+ Mockito.when(meta.extractGraphsFromResponse("new-add")).thenReturn(List.of(key));
+ Mockito.when(meta.extractGraphsFromResponse("old-remove")).thenReturn(List.of(key));
+ Mockito.when(meta.getGraphConfig("DEFAULT", name)).thenAnswer(invocation -> {
+ configReads.incrementAndGet();
+ return missingConfig ? null : new HashMap<>(latest);
+ });
+ Mockito.when(meta.graphConfigs("DEFAULT")).thenReturn(
+ missingConfig ? java.util.Collections.emptyMap() : Map.of(key, latest));
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Whitebox.setInternalState(manager, "pdPeers", "127.0.0.1:8686");
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ spaces.put("DEFAULT", space);
+ // Independent key watches can deliver the new ADD before the old REMOVE.
+ for (int i = 0; i < 2; i++) {
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphAddHandler", manager, "new-add");
+ Assert.assertSame(owner, manager.graph(key));
+ Assert.assertSame(owner, factories.get(key));
+ Assert.assertFalse(owner.closed());
+ Assert.assertEquals(0, configReads.get());
+ Assert.assertEquals(1L, owner.traversal().V().count().next().longValue());
+ owner.tx().close();
+ }
+ if (!removeReceived) {
+ Assert.assertSame(owner, bindings.getGraph(key));
+ return;
+ }
+ String threadName = Thread.currentThread().getName();
+ try {
+ Thread.currentThread().setName("grpc-add-before-remove-control");
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphRemoveHandler", manager, "old-remove");
+ } finally {
+ Thread.currentThread().setName(threadName);
+ }
+ Assert.assertTrue(owner.closed());
+ Mockito.verify(meta, Mockito.never()).removeGraphConfig("DEFAULT", name);
+ Mockito.verify(meta, Mockito.never()).notifyGraphRemove("DEFAULT", name);
+ Assert.assertEquals(5, space.graphNumberUsed());
+ if (missingConfig) {
+ Assert.assertNull(manager.graph(key));
+ Assert.assertFalse(factories.containsKey(key));
+ hub.notifySync(Events.GRAPH_DROP, owner);
+ Assert.assertNull(bindings.getGraph(key));
+ return;
+ }
+ replacement = manager.graph(key);
+ Assert.assertNotNull("The received new ADD was lost before the old REMOVE", replacement);
+ Assert.assertNotSame(owner, replacement);
+ Assert.assertSame(replacement, factories.get(key));
+ Assert.assertSame(replacement, bindings.getGraph(key));
+ Assert.assertSame(replacement, bindings.getAsBindings().get(key));
+ Assert.assertTrue(replacement.started());
+ Assert.assertEquals("new-before-remove", replacement.configuration().get(CoreOptions.ALIAS_NAME));
+ replacement.schema().vertexLabel("new_marker").useAutomaticId().create();
+ replacement.addVertex(org.apache.tinkerpop.gremlin.structure.T.label, "new_marker");
+ replacement.tx().commit();
+ hub.notifySync(Events.GRAPH_DROP, owner);
+ Assert.assertSame(replacement, bindings.getGraph(key));
+ Assert.assertNotNull(bindings.getTraversalSource("__g_" + key));
+ Assert.assertEquals(1L, replacement.traversal().V().count().next().longValue());
+ } finally {
+ manager.close();
+ gremlin.stop().get(30, TimeUnit.SECONDS);
+ if (owner != null) {
+ owner.close();
+ HugeFactory.remove(owner);
+ }
+ if (replacement != null) {
+ replacement.close();
+ HugeFactory.remove(replacement);
+ }
+ }
+ }
+
+ @Test
+ public void testRpcStartupDoesNotRepublishDeletingGraph() throws Exception {
+ this.checkRpcStartupDuringDeletion(false, false);
+ }
+
+ @Test
+ public void testRpcStartupSkipsOwnerAfterRemovalTokenConsumed() throws Exception {
+ this.checkRpcStartupDuringDeletion(true, false);
+ }
+
+ @Test
+ public void testRpcStartupPublishesOtherOwnerDuringDeletion() throws Exception {
+ this.checkRpcStartupDuringDeletion(true, true);
+ }
+
+ private void checkRpcStartupDuringDeletion(boolean consumeToken, boolean keepOther) throws Exception {
+ String name = keepOther ? "startup_retains_peer" :
+ consumeToken ? "startup_consumed_removal" : "startup_deleting_graph";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph owner = null;
+ HugeGraph keeper = null;
+ ExecutorService deletion = Executors.newSingleThreadExecutor(
+ runnable -> new Thread(runnable, "grpc-startup-delete-proof"));
+ CountDownLatch metadataEntered = new CountDownLatch(1);
+ CountDownLatch metadataRelease = new CountDownLatch(1);
+ Future<?> dropped = null;
+ try {
+ if (keepOther) {
+ keeper = manager.createGraphLocal(name + "_peer", sharedMemoryConfig(name + "_peer"));
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", keeper);
+ }
+ owner = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ HugeGraph graph = owner;
+ // The existing Memory feature fixture exercises real cache RPC registration, not shared DML.
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Mockito.doAnswer(invocation -> {
+ metadataEntered.countDown();
+ E.checkState(metadataRelease.await(10, TimeUnit.SECONDS), "metadata proof barrier timed out");
+ return null;
+ }).when(meta).removeGraphConfig("DEFAULT", name);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ spaces.put("DEFAULT", space);
+ dropped = deletion.submit(() -> new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph"));
+ Assert.assertTrue(metadataEntered.await(5, TimeUnit.SECONDS));
+ java.util.Set<?> removing = Whitebox.getInternalState(manager, "removingGraphs");
+ Assert.assertTrue(removing.contains(graph.spaceGraphName()));
+ if (consumeToken) {
+ Mockito.when(meta.extractGraphsFromResponse(Mockito.any())).thenReturn(
+ java.util.Collections.singletonList(graph.spaceGraphName()));
+ Whitebox.invoke(GraphManager.class, "graphRemoveHandler", manager, new Object());
+ Assert.assertFalse(removing.contains(graph.spaceGraphName()));
+ }
+ Assert.assertFalse(graph.closed());
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 0);
+ // Run actual manager startup while the same owner's placement metadata removal is blocked.
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ Assert.assertFalse(dropped.isDone());
+ Assert.assertEquals("RPC startup exported a removing graph during metadata deletion",
+ keepOther ? 2 : 0, server.config().configs().size());
+ if (keepOther) {
+ Assert.assertFalse(keeper.closed());
+ Assert.assertSame(keeper, manager.graph(keeper.spaceGraphName()));
+ }
+ } finally {
+ metadataRelease.countDown();
+ try {
+ if (dropped != null) {
+ dropped.get(20, TimeUnit.SECONDS);
+ }
+ } finally {
+ deletion.shutdown();
+ Assert.assertTrue(deletion.awaitTermination(20, TimeUnit.SECONDS));
+ manager.close();
+ if (owner != null) {
+ owner.close();
+ HugeFactory.remove(owner);
+ }
+ if (keeper != null) {
+ HugeFactory.remove(keeper);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testNonClearingPeerCloseFailureKeepsRpcAtInitialStartup() throws Exception {
+ String name = "peer_close_open_rpc";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph graph = null;
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ CountDownLatch opened = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ try {
+ graph = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ HugeGraph owner = graph;
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", owner);
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 0);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Mockito.when(meta.extractGraphsFromResponse("peer-remove")).thenReturn(List.of(owner.spaceGraphName()));
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Future<?> transaction = worker.submit(() -> {
+ try {
+ owner.tx().open();
+ opened.countDown();
+ E.checkState(release.await(15, TimeUnit.SECONDS), "Foreign transaction release timed out");
+ } catch (InterruptedException failure) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(failure);
+ } finally {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", owner);
+ }
+ });
+ Assert.assertTrue(opened.await(5, TimeUnit.SECONDS));
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphRemoveHandler", manager, "peer-remove");
+ Assert.assertFalse(owner.closed());
+ Assert.assertSame(owner, manager.graph(owner.spaceGraphName()));
+ assertRpcResources(server, client, 0);
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ assertRpcResources(server, client, 2);
+ release.countDown();
+ transaction.get(5, TimeUnit.SECONDS);
+ String threadName = Thread.currentThread().getName();
+ try {
+ Thread.currentThread().setName("grpc-peer-close-retry");
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphRemoveHandler", manager, "peer-remove");
+ } finally {
+ Thread.currentThread().setName(threadName);
+ }
+ Assert.assertTrue(owner.closed());
+ Assert.assertFalse(manager.graphs().contains(owner.spaceGraphName()));
+ assertRpcResources(server, client, 0);
+ Mockito.verify(meta, Mockito.never()).removeGraphConfig(Mockito.anyString(), Mockito.anyString());
+ } finally {
+ release.countDown();
+ worker.shutdown();
+ Assert.assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ manager.close();
+ if (graph != null) {
+ graph.close();
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ @Test
+ public void testNonPdOrdinaryClosedOwnerCannotCertifyBackendClear() throws Exception {
+ GraphManager manager = this.dropManager("non_pd_uncleared_close");
+ HugeGraph owner = null;
+ HugeGraph keeper = null;
+ try {
+ String config = "gremlin.graph=" + HugeFactory.class.getName() + "\nbackend=memory\nstore=";
+ keeper = manager.createGraphLocal("uncleared_keeper", config + "uncleared_keeper\n");
+ owner = manager.createGraphLocal("uncleared_owner", config + "uncleared_owner\n");
+ HugeGraph graph = owner;
+ owner.close();
+ Assert.assertThrows(IllegalStateException.class, () ->
+ new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", "uncleared_owner", "I'm sure to drop the graph"));
+ Map<String, HugeGraph> factories = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertSame(owner, factories.get(owner.spaceGraphName()));
+ Assert.assertSame(owner, manager.graph(owner.spaceGraphName()));
+ Assert.assertTrue(graph.configuration().file().isFile());
+ } finally {
+ manager.close();
+ if (owner != null) {
+ owner.close();
+ HugeFactory.remove(owner);
+ }
+ if (keeper != null) {
+ HugeFactory.remove(keeper);
+ }
+ }
+ }
+
+ @Test
+ public void testRpcStartupSkipsOrdinaryClosedOwner() throws Exception {
+ GraphManager manager = this.dropManager("startup_closed_owner");
+ HugeGraph owner = null;
+ try {
+ owner = manager.createGraphLocal("startup_closed", sharedMemoryConfig("startup_closed"));
+ owner.close();
+ RuntimeException startup = null;
+ try {
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ } catch (RuntimeException failure) {
+ startup = failure;
+ }
+ Assert.assertNull("RPC startup must skip a closed owner: " + startup, startup);
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 0);
+ } finally {
+ manager.close();
+ if (owner != null) {
+ owner.close();
+ HugeFactory.remove(owner);
+ }
+ }
+ }
+
+ @Test
+ public void testNonPdFailedClearPreservesExistingRpc() throws Exception {
+ GraphManager manager = this.dropManager("non_pd_clear_failure");
+ HugeGraph owner = null;
+ HugeGraph keeper = null;
+ BackendStoreProvider provider = null;
+ try {
+ owner = manager.createGraphLocal("non_pd_clear", sharedMemoryConfig("non_pd_clear"));
+ keeper = manager.createGraphLocal("non_pd_clear_peer", sharedMemoryConfig("non_pd_clear_peer"));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 4);
+ provider = Mockito.spy(owner.storeProvider());
+ Whitebox.setInternalState(owner, "storeProvider", provider);
+ IllegalStateException failure = new IllegalStateException("clear rejected before destructive action");
+ Mockito.doThrow(failure).when(provider).clear();
+ Assert.assertThrows(IllegalStateException.class, () ->
+ new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", "non_pd_clear", "I'm sure to drop the graph"),
+ caught -> Assert.assertSame(failure, caught));
+ Assert.assertFalse(owner.closed());
+ Assert.assertSame(owner, manager.graph(owner.spaceGraphName()));
+ Assert.assertTrue(owner.configuration().file().isFile());
+ java.util.Set<?> pending = Whitebox.getInternalState(manager, "rpcRemovingOwners");
+ Assert.assertFalse(pending.contains(owner));
+ assertRpcResources(server, client, 4);
+ } finally {
+ if (provider != null) {
+ Mockito.doCallRealMethod().when(provider).clear();
+ }
+ manager.close();
+ if (owner != null) {
+ HugeFactory.remove(owner);
+ }
+ if (keeper != null) {
+ HugeFactory.remove(keeper);
+ }
+ }
+ }
+
+ @Test
+ public void testNonPdClearFailureDoesNotUnregisterRpcDuringStartup() throws Exception {
+ String name = "non_pd_clear_startup";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph owner = null;
+ HugeGraph keeper = null;
+ BackendStoreProvider provider = null;
+ CountDownLatch clearEntered = new CountDownLatch(1);
+ CountDownLatch clearRelease = new CountDownLatch(1);
+ ExecutorService deletion = Executors.newSingleThreadExecutor();
+ Future<?> dropped = null;
+ try {
+ owner = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ keeper = manager.createGraphLocal(name + "_peer", sharedMemoryConfig(name + "_peer"));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", owner);
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", keeper);
+ provider = Mockito.spy(owner.storeProvider());
+ Whitebox.setInternalState(owner, "storeProvider", provider);
+ IllegalStateException failure = new IllegalStateException("clear fails before destructive action");
+ Mockito.doAnswer(invocation -> {
+ clearEntered.countDown();
+ E.checkState(clearRelease.await(10, TimeUnit.SECONDS), "local clear proof barrier timed out");
+ throw failure;
+ }).when(provider).clear();
+ dropped = deletion.submit(() -> new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph"));
+ Assert.assertTrue(clearEntered.await(5, TimeUnit.SECONDS));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ clearRelease.countDown();
+ Future<?> failedDrop = dropped;
+ Assert.assertThrows(ExecutionException.class, () -> failedDrop.get(20, TimeUnit.SECONDS),
+ caught -> Assert.assertSame(failure, caught.getCause()));
+ dropped = null;
+ Assert.assertFalse(owner.closed());
+ Assert.assertSame(owner, manager.graph(owner.spaceGraphName()));
+ Assert.assertTrue(owner.configuration().file().isFile());
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 4);
+ } finally {
+ clearRelease.countDown();
+ try {
+ if (dropped != null) {
+ try {
+ dropped.get(20, TimeUnit.SECONDS);
+ } catch (ExecutionException expected) {
+ // The controlled clear fails before destructive work.
+ }
+ }
+ } finally {
+ if (provider != null) {
+ Mockito.doCallRealMethod().when(provider).clear();
+ }
+ deletion.shutdown();
+ Assert.assertTrue(deletion.awaitTermination(20, TimeUnit.SECONDS));
+ manager.close();
+ if (owner != null) {
+ HugeFactory.remove(owner);
+ }
+ if (keeper != null) {
+ HugeFactory.remove(keeper);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testMetadataFailureRestoresRpcBeforeClearingRemovalMarker() throws Exception {
+ this.checkMetadataRpcRestoration(false);
+ }
+
+ @Test
+ public void testFailedRpcRestorationRetainsRemovalOwner() throws Exception {
+ this.checkMetadataRpcRestoration(true);
+ }
+
+ private void checkMetadataRpcRestoration(boolean rejectRestore) throws Exception {
+ String name = rejectRestore ? "metadata_restore_failure" : "metadata_restore_success";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph owner = null;
+ RpcServer server = null;
+ try {
+ owner = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ RpcServer original = Whitebox.getInternalState(manager, "rpcServer");
+ server = Mockito.spy(original);
+ Whitebox.setInternalState(manager, "rpcServer", server);
+ IllegalStateException metadata = new IllegalStateException("placement removal failed");
+ IllegalStateException restoration = new IllegalStateException("RPC restoration failed");
+ if (rejectRestore) {
+ Mockito.doThrow(restoration).when(server).exportAll();
+ }
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Mockito.doThrow(metadata).when(meta).removeGraphConfig("DEFAULT", name);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ spaces.put("DEFAULT", space);
+ Assert.assertThrows(HugeException.class, () ->
+ new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph"),
+ caught -> Assert.assertSame(metadata, caught.getCause()));
+ java.util.Set<?> pending = Whitebox.getInternalState(manager, "rpcRemovingOwners");
+ Assert.assertEquals(rejectRestore, pending.contains(owner));
+ Assert.assertFalse(owner.closed());
+ Assert.assertSame(owner, manager.graph(owner.spaceGraphName()));
+ Assert.assertTrue(owner.configuration().file().isFile());
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, rejectRestore ? 0 : 2);
+ if (rejectRestore) {
+ Assert.assertEquals(1, metadata.getSuppressed().length);
+ Assert.assertSame(restoration, metadata.getSuppressed()[0]);
+ }
+ } finally {
+ if (server != null) {
+ Mockito.doCallRealMethod().when(server).exportAll();
+ }
+ manager.close();
+ if (owner != null) {
+ HugeFactory.remove(owner);
+ }
+ }
+ }
+
+ @Test
+ public void testStartupDoesNotExportPendingOwnerAfterUnexportFailure() throws Exception {
+ String name = "startup_unexport_failure";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph owner = null;
+ RpcServer server = null;
+ com.alipay.sofa.rpc.config.ProviderConfig<?> retained = null;
+ try {
+ owner = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ RpcServer original = Whitebox.getInternalState(manager, "rpcServer");
+ server = Mockito.spy(original);
+ Whitebox.setInternalState(manager, "rpcServer", server);
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ owner.registerRpcServices(server.config(), client.config());
+ assertRpcResources(server, client, 2);
+ Map<String, com.alipay.sofa.rpc.config.ProviderConfig<?>> configs = server.config().configs();
+ Map.Entry<String, com.alipay.sofa.rpc.config.ProviderConfig<?>> entry =
+ configs.entrySet().iterator().next();
+ retained = Mockito.spy(entry.getValue());
+ configs.put(entry.getKey(), retained);
+ IllegalStateException failure = new IllegalStateException("provider unexport rejected");
+ Mockito.doThrow(failure).when(retained).unExport();
+ Whitebox.setInternalState(manager, "PDExist", true);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Assert.assertThrows(IllegalStateException.class, () ->
+ new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph"));
+ Assert.assertEquals(1, server.config().configs().size());
+ Mockito.verify(meta, Mockito.never()).removeGraphConfig(Mockito.any(), Mockito.any());
+ Assert.assertThrows(IllegalStateException.class, () ->
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager));
+ Mockito.verify(server, Mockito.never()).exportAll();
+ Assert.assertFalse(owner.closed());
+ Assert.assertSame(owner, manager.graph(owner.spaceGraphName()));
+ java.util.Set<?> pending = Whitebox.getInternalState(manager, "rpcRemovingOwners");
+ Assert.assertTrue(pending.contains(owner));
+ } finally {
+ if (retained != null) {
+ Mockito.doCallRealMethod().when(retained).unExport();
+ }
+ manager.close();
+ if (owner != null) {
+ owner.close();
+ HugeFactory.remove(owner);
+ }
+ }
+ }
+
+ @Test
+ public void testNonPdHealthyPublicDeleteCompletesCleanup() throws Exception {
+ this.checkNonPdPublicDeleteRetry(false, false);
+ }
+
+ @Test
+ public void testNonPdPublicDeleteRetriesFailedProviderCleanup() throws Exception {
+ this.checkNonPdPublicDeleteRetry(true, false);
+ }
+
+ @Test
+ public void testNonPdPublicDeleteRetriesFailedRpcUnregistration() throws Exception {
+ this.checkNonPdPublicDeleteRetry(true, true);
+ }
+
+ @Test
+ public void testNonPdPublicDeletePreservesFatalCleanupIdentity() throws Exception {
+ this.checkNonPdPublicDeleteRetry(true, false, true);
+ }
+
+ private void checkNonPdPublicDeleteRetry(boolean failCleanup, boolean rpcCleanup) throws Exception {
+ this.checkNonPdPublicDeleteRetry(failCleanup, rpcCleanup, false);
+ }
+
+ private void checkNonPdPublicDeleteRetry(boolean failCleanup, boolean rpcCleanup, boolean fatal) throws Exception {
+ String name = !failCleanup ? "non_pd_healthy" : rpcCleanup ? "non_pd_rpc_retry" :
+ fatal ? "non_pd_fatal_retry" : "non_pd_provider_retry";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph owner = null;
+ HugeGraph keeper = null;
+ AtomicBoolean reject = new AtomicBoolean(failCleanup);
+ AtomicInteger attempts = new AtomicInteger();
+ try {
+ String config = "gremlin.graph=" + HugeFactory.class.getName() + "\nbackend=memory\nstore=";
+ keeper = manager.createGraphLocal(name + "_keeper", config + name + "_keeper\n");
+ owner = manager.createGraphLocal(name, config + name + "\n");
+ HugeGraph graph = owner;
+ Map<String, HugeGraph> factories = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Map<String, HugeGraph> managed = Whitebox.getInternalState(manager, "graphs");
+ List<?> retained = Whitebox.getInternalState(owner, "closeResources");
+ List<Runnable> rpc = Whitebox.getInternalState(owner, "rpcServiceCleanup");
+ File file = owner.configuration().file();
+ Assert.assertSame(owner, factories.get(owner.spaceGraphName()));
+ Assert.assertSame(owner, managed.get(owner.spaceGraphName()));
+ Assert.assertTrue(file.isFile());
+ Throwable controlled = fatal ? new AssertionError("fatal native cleanup boundary") :
+ new IllegalStateException("persistent native cleanup boundary");
+ BackendStoreProvider provider = Mockito.spy(owner.storeProvider());
+ Whitebox.setInternalState(owner, "storeProvider", provider);
+ AtomicInteger clears = new AtomicInteger();
+ Mockito.doAnswer(invocation -> {
+ clears.incrementAndGet();
+ invocation.callRealMethod();
+ return null;
+ }).when(provider).clear();
+ if (rpcCleanup) {
+ // Only the RPC cleanup callback is controlled; actual graph close/unregister queues run.
+ rpc.add(() -> {
+ attempts.incrementAndGet();
+ if (reject.get()) {
+ throw (RuntimeException) controlled;
+ }
+ });
+ } else {
+ Mockito.doAnswer(invocation -> {
+ invocation.callRealMethod();
+ attempts.incrementAndGet();
+ if (reject.get()) {
+ throw controlled;
+ }
+ return null;
+ }).when(provider).close();
+ }
+ org.apache.hugegraph.api.profile.GraphsAPI api = new org.apache.hugegraph.api.profile.GraphsAPI();
+ Throwable first = null;
+ try {
+ api.drop(manager, "DEFAULT", name, "I'm sure to drop the graph");
+ } catch (RuntimeException | Error failure) {
+ first = failure;
+ }
+ boolean configRetained = file.isFile();
+ boolean factoryRetained = factories.get(owner.spaceGraphName()) == owner;
+ boolean managerRetained = managed.get(owner.spaceGraphName()) == owner;
+ boolean completed = Whitebox.getInternalState(owner, "closeCompleted");
+ int pending = retained.size();
+ int rpcPending = rpc.size();
+ int firstAttempts = attempts.get();
+ reject.set(false);
+ Throwable retry = null;
+ if (failCleanup) {
+ try {
+ api.drop(manager, "DEFAULT", name, "I'm sure to drop the graph");
+ } catch (RuntimeException | Error failure) {
+ retry = failure;
+ }
+ }
+ String facts = String.format("native non-PD DELETE: rpc=%s first=%s retry=%s closed=%s " +
+ "closeCompleted=%s config=%s factory=%s manager=%s retained=%s rpcPending=%s " +
+ "attemptsBeforeRetry=%s attemptsAfterRetry=%s", rpcCleanup, first, retry, graph.closed(),
+ completed, configRetained, factoryRetained, managerRetained, pending, rpcPending,
+ firstAttempts, attempts.get());
+ if (failCleanup) {
+ Assert.assertTrue(facts, first != null);
+ if (fatal) {
+ Assert.assertSame(controlled, first);
+ }
+ Assert.assertTrue(facts, configRetained);
+ Assert.assertTrue(facts, factoryRetained);
+ Assert.assertTrue(facts, managerRetained);
+ Assert.assertFalse(facts, completed);
+ Assert.assertTrue(facts, pending > 0);
+ Assert.assertNull(facts, retry);
+ Assert.assertTrue(facts, attempts.get() > firstAttempts);
+ } else {
+ Assert.assertNull(facts, first);
+ Assert.assertTrue(facts, completed);
+ Assert.assertEquals(facts, 0, pending);
+ }
+ Assert.assertEquals("A successful drop clear must not repeat during cleanup retry", 1, clears.get());
+ Assert.assertTrue(Whitebox.getInternalState(owner, "closeCompleted"));
+ Assert.assertEquals(0, retained.size());
+ Assert.assertFalse(file.exists());
+ Assert.assertFalse(factories.containsKey(owner.spaceGraphName()));
+ Assert.assertFalse(managed.containsKey(owner.spaceGraphName()));
+ Assert.assertSame(keeper, manager.graph(keeper.spaceGraphName()));
+ } finally {
+ reject.set(false);
+ if (owner != null) {
+ owner.close();
+ HugeFactory.remove(owner);
+ }
+ manager.close();
+ if (keeper != null) {
+ HugeFactory.remove(keeper);
+ }
+ }
+ }
+
+ private GraphManager dropManager(String name) throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder(name).getPath());
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false);
+ properties.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "127.0.0.1");
+ properties.setProperty(RpcOptions.RPC_SERVER_PORT.name(), 0);
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "127.0.0.1:1");
+ return new GraphManager(new HugeConfig(properties), new EventHub(name));
+ }
+
+ private void checkPostCloseDrop(boolean failCleanup) throws Exception {
+ String name = failCleanup ? "closed_rpc_retry" : "closed_failure";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph owner = null;
+ try {
+ owner = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ HugeGraph delegate = owner;
+ Map<String, HugeGraph> factoryOwners = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ factoryOwners.put(owner.spaceGraphName(), owner);
+ IllegalStateException failure = new IllegalStateException("closed owner cleanup failed");
+ AtomicInteger clears = new AtomicInteger();
+ AtomicInteger rpcCleanup = new AtomicInteger();
+ BackendStoreProvider provider = Mockito.spy(owner.storeProvider());
+ if (!failCleanup) {
+ Whitebox.setInternalState(owner, "storeProvider", provider);
+ Mockito.doAnswer(invocation -> {
+ invocation.callRealMethod();
+ throw failure;
+ }).when(provider).close();
+ }
+ HugeGraph graph = (HugeGraph) Proxy.newProxyInstance(HugeGraph.class.getClassLoader(),
+ new Class<?>[]{HugeGraph.class}, (proxy, method, args) -> {
+ if ("clearBackend".equals(method.getName())) {
+ clears.incrementAndGet();
+ }
+ if ("unregisterRpcServices".equals(method.getName()) &&
+ rpcCleanup.incrementAndGet() == 2 && failCleanup) {
+ throw failure;
+ }
+ try {
+ return method.invoke(delegate, args);
+ } catch (InvocationTargetException error) {
+ throw error.getCause();
+ }
+ });
+ Map<String, HugeGraph> graphs = Whitebox.getInternalState(manager, "graphs");
+ graphs.put(graph.spaceGraphName(), graph);
+ DefaultGraphManager bindings = new DefaultGraphManager(new Settings());
+ bindings.putGraph(graph.spaceGraphName(), graph);
+ EventHub hub = Whitebox.getInternalState(manager, "eventHub");
+ CountDownLatch dropped = new CountDownLatch(1);
+ hub.listen(Events.GRAPH_DROP, event -> {
+ try {
+ bindings.removeGraph(graph.spaceGraphName());
+ } catch (Exception failureToUnbind) {
+ throw new IllegalStateException(failureToUnbind);
+ }
+ dropped.countDown();
+ return true;
+ });
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ spaces.put("DEFAULT", space);
+ if (failCleanup) {
+ Assert.assertThrows(IllegalStateException.class, () -> manager.dropGraph(
+ "DEFAULT", name, true), caught -> Assert.assertSame(failure, caught));
+ } else {
+ Assert.assertThrows(HugeException.class, () -> manager.dropGraph(
+ "DEFAULT", name, true), caught -> Assert.assertSame(failure, caught.getCause()));
+ Mockito.doCallRealMethod().when(provider).close();
+ }
+ Assert.assertTrue(owner.closed());
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ Assert.assertSame(graph, bindings.getGraph(graph.spaceGraphName()));
+ Map<?, ?> retainedFactory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertSame(owner, retainedFactory.get(graph.spaceGraphName()));
+ Assert.assertEquals(5, space.graphNumberUsed());
+ Assert.assertEquals(1L, dropped.getCount());
+ manager.dropGraph("DEFAULT", name, true);
+ Assert.assertTrue(dropped.await(5, TimeUnit.SECONDS));
+ Assert.assertTrue(owner.closed());
+ Assert.assertEquals(1, clears.get());
+ Mockito.verify(meta).removeGraphConfig("DEFAULT", name);
+ Mockito.verify(meta).notifyGraphRemove("DEFAULT", name);
+ Mockito.verify(meta).updateGraphSpaceConfig("DEFAULT", space);
+ Assert.assertEquals(4, space.graphNumberUsed());
+ Assert.assertNull(bindings.getGraph(graph.spaceGraphName()));
+ Assert.assertFalse(manager.graphs().contains(graph.spaceGraphName()));
+ Map<?, ?> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertFalse(factory.containsKey(graph.spaceGraphName()));
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 0);
+ } finally {
+ manager.close();
+ if (owner != null) {
+ HugeFactory.remove(owner);
+ }
+ }
+ }
+
+ @Test
+ public void testQuotaUpdateFailureWithClearRetriesClosedOwner() throws Exception {
+ this.checkLegacyDrop(false, true, true);
+ }
+
+ @Test
+ public void testRaftCloseFailuresKeepOwnerAndPreserveSuppressed() throws Exception {
+ this.checkRaftCloseFailures(false);
+ }
+
+ @Test
+ public void testRaftCloseFailuresDoNotSuppressTheSameException() throws Exception {
+ this.checkRaftCloseFailures(true);
+ }
+
+ @Test
+ public void testRaftClosePrioritizesFatalError() throws Exception {
+ this.checkRaftCloseFailures(false, true);
+ }
+
+ private void checkRaftCloseFailures(boolean sameFailure) throws Exception {
+ this.checkRaftCloseFailures(sameFailure, false);
+ }
+
+ private void checkRaftCloseFailures(boolean sameFailure, boolean fatalNode) throws Exception {
+ String name = fatalNode ? "raft_fatal_node" : sameFailure ? "raft_same_failure" : "raft_close_failures";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph graph = null;
+ com.alipay.sofa.jraft.rpc.RpcServer nativeRpc = null;
+ RaftContext context = null;
+ String threadName = Thread.currentThread().getName();
+ AtomicBoolean failProvider = new AtomicBoolean(true);
+ AtomicBoolean failNode = new AtomicBoolean(true);
+ AtomicBoolean failRpc = new AtomicBoolean(true);
+ try {
+ graph = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ Map<String, HugeGraph> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ // The shared-memory GraphFactory fixture bypasses HugeFactory.open().
+ factory.put(graph.spaceGraphName(), graph);
+ nativeRpc = startNativeRaftRpc();
+ int port = nativeRpc.boundPort();
+ Assert.assertTrue(nativeRpcListening(port));
+ graph.configuration().setProperty("raft.group_peers", "127.0.0.1:" + port);
+ BackendStoreProvider provider = Mockito.spy(graph.storeProvider());
+ IllegalStateException primary = new IllegalStateException("provider failed before release");
+ Throwable nodeFailure = fatalNode ? new AssertionError("fatal node close") : sameFailure ? primary :
+ new IllegalStateException("node failed before release");
+ IllegalStateException rpcFailure = sameFailure ? primary :
+ new IllegalStateException("RPC failed before release");
+ AtomicInteger providerCalls = new AtomicInteger();
+ AtomicInteger nodeCalls = new AtomicInteger();
+ AtomicInteger rpcCalls = new AtomicInteger();
+ Mockito.doAnswer(call -> {
+ providerCalls.incrementAndGet();
+ if (failProvider.get()) {
+ throw primary;
+ }
+ return call.callRealMethod();
+ }).when(provider).close();
+ HugeGraphParams params = Whitebox.getInternalState(graph, "params");
+ RaftBackendStoreProvider raft = new RaftBackendStoreProvider(params, provider);
+ context = Whitebox.getInternalState(raft, "context");
+ RaftNode node = Mockito.mock(RaftNode.class);
+ Mockito.doAnswer(call -> {
+ nodeCalls.incrementAndGet();
+ if (failNode.get()) {
+ throw nodeFailure;
+ }
+ return null;
+ }).when(node).shutdown();
+ com.alipay.sofa.jraft.rpc.RpcServer rpc = Mockito.spy(nativeRpc);
+ Mockito.doAnswer(call -> {
+ rpcCalls.incrementAndGet();
+ if (failRpc.get()) {
+ throw rpcFailure;
+ }
+ return call.callRealMethod();
+ }).when(rpc).shutdown();
+ Whitebox.setInternalState(context, "raftNode", node);
+ Whitebox.setInternalState(context, "raftRpcServer", rpc);
+ Whitebox.setInternalState(context, "endpoint", new PeerId("127.0.0.1", port));
+ Whitebox.setInternalState(graph, "storeProvider", raft);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Thread.currentThread().setName("grpc-raft-close-failures");
+ if (fatalNode) {
+ AssertionError fatal = Assert.assertThrows(AssertionError.class,
+ () -> manager.dropGraph("DEFAULT", name, false));
+ Assert.assertSame(nodeFailure, fatal);
+ Assert.assertEquals(2, fatal.getSuppressed().length);
+ Assert.assertSame(rpcFailure, fatal.getSuppressed()[0]);
+ Assert.assertSame(primary, fatal.getSuppressed()[1]);
+ } else {
+ HugeException failure = Assert.assertThrows(HugeException.class,
+ () -> manager.dropGraph("DEFAULT", name, false));
+ Assert.assertSame(primary, failure.getCause());
+ Assert.assertEquals(sameFailure ? 0 : 1, primary.getSuppressed().length);
+ if (!sameFailure) {
+ Assert.assertSame(nodeFailure, primary.getSuppressed()[0]);
+ Assert.assertSame(rpcFailure, nodeFailure.getSuppressed()[0]);
+ }
+ }
+ Assert.assertTrue(graph.closed());
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ Assert.assertSame(graph, factory.get(graph.spaceGraphName()));
+ Assert.assertTrue(nativeRpcListening(port));
+ Assert.assertFalse(context.snapshotExecutor().isShutdown());
+ Assert.assertFalse(context.backendExecutor().isShutdown());
+ failProvider.set(false);
+ failNode.set(false);
+ // A remaining listener failure must retain the logically closed owner too.
+ // Retain the original non-clearing removal contract on every retry.
+ HugeException failure = Assert.assertThrows(HugeException.class,
+ () -> manager.dropGraph("DEFAULT", name, false));
+ Assert.assertSame(rpcFailure, failure.getCause());
+ assertRaftExecutorsTerminated(context);
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ Assert.assertSame(graph, factory.get(graph.spaceGraphName()));
+ Assert.assertTrue(nativeRpcListening(port));
+ failRpc.set(false);
+ manager.dropGraph("DEFAULT", name, false);
+ Assert.assertFalse(manager.graphs().contains(graph.spaceGraphName()));
+ Assert.assertFalse(factory.containsKey(graph.spaceGraphName()));
+ Assert.assertEquals(2, providerCalls.get());
+ Assert.assertEquals(2, nodeCalls.get());
+ Assert.assertEquals(3, rpcCalls.get());
+ Mockito.verify(meta, Mockito.never()).removeGraphConfig("DEFAULT", name);
+ Mockito.verify(meta, Mockito.never()).notifyGraphRemove("DEFAULT", name);
+ Mockito.verify(meta, Mockito.never()).updateGraphSpaceConfig(Mockito.eq("DEFAULT"),
+ Mockito.any(GraphSpace.class));
+ awaitNativeRpcClosed(port);
+ } finally {
+ failProvider.set(false);
+ failNode.set(false);
+ failRpc.set(false);
+ Thread.currentThread().setName(threadName);
+ if (nativeRpc != null && nativeRpcListening(nativeRpc.boundPort())) {
+ nativeRpc.shutdown();
+ }
+ stopRaftFixtureExecutors(context);
+ manager.close();
+ if (graph != null) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ @Test
+ public void testProviderFailureClosesRealRaftNodeBeforeOwnerRemoval() throws Exception {
+ OptionSpace.register("rocksdb", RocksDBOptions.instance());
+ BackendProviderFactory.register("rocksdb", RocksDBStoreProvider.class.getName());
+ String name = "raft_real_close";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph graph = null;
+ com.alipay.sofa.jraft.rpc.RpcServer rpc = null;
+ RaftContext context = null;
+ RaftNode raftNode = null;
+ String threadName = Thread.currentThread().getName();
+ try {
+ String data = this.temporary.newFolder("raft_data").getPath();
+ String wal = this.temporary.newFolder("raft_wal").getPath();
+ graph = manager.createGraphLocal(name, "gremlin.graph=" + HugeFactory.class.getName() +
+ "\nbackend=rocksdb\nserializer=binary\nstore=" + name +
+ "\nrocksdb.data_path=" + data + "\nrocksdb.wal_path=" + wal + "\n");
+ rpc = startNativeRaftRpc();
+ int port = rpc.boundPort();
+ Assert.assertTrue(nativeRpcListening(port));
+ HugeConfig config = graph.configuration();
+ config.setProperty("raft.group_peers", "127.0.0.1:" + port);
+ config.setProperty(CoreOptions.RAFT_PATH.name(), this.temporary.newFolder("raft_log").getPath());
+ config.setProperty(CoreOptions.RAFT_ELECTION_TIMEOUT.name(), 300);
+ config.setProperty(CoreOptions.RAFT_SNAPSHOT_INTERVAL.name(), 0);
+ config.setProperty(CoreOptions.RAFT_RPC_THREADS.name(), 2);
+ config.setProperty(CoreOptions.RAFT_QUEUE_SIZE.name(), 2048);
+ config.setProperty(CoreOptions.RAFT_SAFE_READ.name(), true);
+ BackendStoreProvider provider = Mockito.spy(graph.storeProvider());
+ AtomicInteger providerCalls = new AtomicInteger();
+ IllegalStateException primary = new IllegalStateException("provider failed before real node release");
+ Mockito.doAnswer(call -> {
+ providerCalls.incrementAndGet();
+ throw primary;
+ }).when(provider).close();
+ HugeGraphParams params = Whitebox.getInternalState(graph, "params");
+ RaftBackendStoreProvider raft = new RaftBackendStoreProvider(params, provider);
+ raft.loadSchemaStore(config);
+ raft.loadGraphStore(config);
+ raft.loadSystemStore(config);
+ context = Whitebox.getInternalState(raft, "context");
+ Whitebox.setInternalState(context, "raftRpcServer", rpc);
+ Whitebox.setInternalState(context, "endpoint", new PeerId("127.0.0.1", port));
+ NodeManager.getInstance().addAddress(new Endpoint("127.0.0.1", port));
+ raftNode = new RaftNode(context);
+ Whitebox.setInternalState(context, "raftNode", raftNode);
+ Node node = Whitebox.invoke(RaftNode.class, "node", raftNode);
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (node.getNodeState() != State.STATE_LEADER && System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ }
+ Assert.assertEquals(State.STATE_LEADER, node.getNodeState());
+ Whitebox.setInternalState(graph, "storeProvider", raft);
+ Whitebox.setInternalState(manager, "metaManager", Mockito.mock(MetaManager.class));
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Thread.currentThread().setName("grpc-real-raft-close");
+ HugeException failure = Assert.assertThrows(HugeException.class,
+ () -> manager.dropGraph("DEFAULT", name, false));
+ Assert.assertSame(primary, failure.getCause());
+ Assert.assertTrue(graph.closed());
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ Map<?, ?> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertSame(graph, factory.get(graph.spaceGraphName()));
+ Assert.assertEquals(State.STATE_SHUTDOWN, node.getNodeState());
+ assertRaftExecutorsTerminated(context);
+ awaitNativeRpcClosed(port);
+ try (ServerSocket reused = new ServerSocket(port)) {
+ Assert.assertEquals(port, reused.getLocalPort());
+ }
+ Mockito.doAnswer(call -> {
+ providerCalls.incrementAndGet();
+ return call.callRealMethod();
+ }).when(provider).close();
+ manager.dropGraph("DEFAULT", name, false);
+ Assert.assertEquals(2, providerCalls.get());
+ Assert.assertFalse(manager.graphs().contains(graph.spaceGraphName()));
+ Assert.assertFalse(factory.containsKey(graph.spaceGraphName()));
+ Assert.assertEquals(State.STATE_SHUTDOWN, node.getNodeState());
+ } finally {
+ Thread.currentThread().setName(threadName);
+ if (raftNode != null) {
+ Node node = Whitebox.invoke(RaftNode.class, "node", raftNode);
+ if (node.getNodeState() != State.STATE_SHUTDOWN) {
+ raftNode.shutdown();
+ }
+ }
+ if (rpc != null && nativeRpcListening(rpc.boundPort())) {
+ rpc.shutdown();
+ }
+ stopRaftFixtureExecutors(context);
+ manager.close();
+ if (graph != null) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ @Test
+ public void testSharedRaftListenerSurvivesGraphDropAndOwnerRestart() throws Exception {
+ GraphManager first = this.raftOwnerManager("shared_owner_one", 0);
+ GraphManager second = null;
+ List<HugeGraph> graphs = new ArrayList<>();
+ List<RaftContext> contexts = new ArrayList<>();
+ CliClientServiceImpl client = new CliClientServiceImpl();
+ client.init(new CliOptions());
+ String threadName = Thread.currentThread().getName();
+ try {
+ com.alipay.remoting.rpc.RpcServer nativeFirst = Whitebox.invoke(GraphManager.class,
+ "remotingRpcServer", first);
+ int port = nativeFirst.port();
+ Endpoint address = new Endpoint(nativeFirst.ip(), port);
+ Endpoint connect = new Endpoint("127.0.0.1", port);
+ HugeGraph a = this.attachRaftGraph(first, "shared_real_a", nativeFirst, contexts);
+ HugeGraph b = this.attachRaftGraph(first, "shared_real_b", nativeFirst, contexts);
+ graphs.add(a);
+ graphs.add(b);
+ startSharedRaftPair(contexts, nativeFirst);
+ Node nodeA = nativeNode(contexts.get(0));
+ Node nodeB = nativeNode(contexts.get(1));
+ assertRaftLeader(client, connect, b.spaceGraphName());
+ Whitebox.setInternalState(first, "metaManager", Mockito.mock(MetaManager.class));
+ Whitebox.setInternalState(first, "PDExist", true);
+ Thread.currentThread().setName("grpc-shared-real-owner");
+ first.dropGraph("DEFAULT", a.name(), false);
+ Assert.assertEquals(State.STATE_SHUTDOWN, nodeA.getNodeState());
+ assertRaftExecutorsTerminated(contexts.get(0));
+ Assert.assertEquals(State.STATE_LEADER, nodeB.getNodeState());
+ Assert.assertTrue(nativeRpcListening(port));
+ Assert.assertTrue(NodeManager.getInstance().serverExists(address));
+ assertRaftLeader(client, connect, b.spaceGraphName());
+ client.shutdown();
+ client = null;
+ RpcServer server = Mockito.spy((RpcServer) Whitebox.getInternalState(first, "rpcServer"));
+ IllegalStateException destroyFailure = new IllegalStateException("owner listener release failed");
+ Mockito.doThrow(destroyFailure).when(server).destroy();
+ Whitebox.setInternalState(first, "rpcServer", server);
+ Assert.assertThrows(IllegalStateException.class, first::close,
+ error -> Assert.assertSame(destroyFailure, error));
+ Assert.assertEquals(State.STATE_SHUTDOWN, nodeB.getNodeState());
+ Assert.assertTrue(nativeRpcListening(port));
+ Assert.assertTrue(NodeManager.getInstance().serverExists(address));
+ Assert.assertEquals(address, Whitebox.getInternalState(first, "raftRpcEndpoint"));
+ Mockito.doCallRealMethod().when(server).destroy();
+ first.close();
+ Mockito.verify(server, Mockito.times(2)).destroy();
+ Assert.assertNull(Whitebox.getInternalState(first, "raftRpcEndpoint"));
+ Assert.assertEquals(State.STATE_SHUTDOWN, nodeB.getNodeState());
+ awaitNativeRpcClosed(port);
+ Assert.assertFalse(NodeManager.getInstance().serverExists(address));
+ List<Node> stopped = NodeManager.getInstance().getNodesByGroupId(b.spaceGraphName());
+ Assert.assertTrue(stopped == null || stopped.isEmpty());
+ try (ServerSocket reused = new ServerSocket(port)) {
+ Assert.assertEquals(port, reused.getLocalPort());
+ }
+ second = this.raftOwnerManager("shared_owner_two", port);
+ com.alipay.remoting.rpc.RpcServer nativeSecond = Whitebox.invoke(GraphManager.class,
+ "remotingRpcServer", second);
+ Assert.assertNotSame(nativeFirst, nativeSecond);
+ Assert.assertEquals(port, nativeSecond.port());
+ HugeGraph restarted = this.attachRaftGraph(second, b.name(), nativeSecond, contexts);
+ graphs.add(restarted);
+ contexts.get(2).initRaftNode(nativeSecond);
+ Node fresh = nativeNode(contexts.get(2));
+ awaitRaftLeader(fresh);
+ Assert.assertNotSame(nodeB, fresh);
+ client = new CliClientServiceImpl();
+ client.init(new CliOptions());
+ assertRaftLeader(client, connect, restarted.spaceGraphName());
+ client.shutdown();
+ client = null;
+ second.close();
+ Assert.assertEquals(State.STATE_SHUTDOWN, fresh.getNodeState());
+ assertRaftExecutorsTerminated(contexts.get(2));
+ awaitNativeRpcClosed(port);
+ Assert.assertFalse(NodeManager.getInstance().serverExists(address));
+ } finally {
+ Thread.currentThread().setName(threadName);
+ if (client != null) {
+ client.shutdown();
+ }
+ first.close();
+ if (second != null) {
+ second.close();
+ }
+ for (RaftContext context : contexts) {
+ stopRaftFixtureExecutors(context);
+ }
+ for (HugeGraph graph : graphs) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ @Test
+ public void testInterruptedRaftJoinRetainsOwnerUntilRetry() throws Exception {
+ GraphManager manager = this.raftOwnerManager("interrupt_shared_owner", 0);
+ List<HugeGraph> graphs = new ArrayList<>();
+ List<RaftContext> contexts = new ArrayList<>();
+ CliClientServiceImpl client = new CliClientServiceImpl();
+ client.init(new CliOptions());
+ String threadName = Thread.currentThread().getName();
+ try {
+ com.alipay.remoting.rpc.RpcServer shared = Whitebox.invoke(GraphManager.class,
+ "remotingRpcServer", manager);
+ Endpoint address = new Endpoint(shared.ip(), shared.port());
+ Endpoint connect = new Endpoint("127.0.0.1", shared.port());
+ HugeGraph a = this.attachRaftGraph(manager, "interrupt_real_a", shared, contexts);
+ HugeGraph b = this.attachRaftGraph(manager, "interrupt_real_b", shared, contexts);
+ graphs.add(a);
+ graphs.add(b);
+ startSharedRaftPair(contexts, shared);
+ RaftContext context = contexts.get(0);
+ Node nodeA = nativeNode(context);
+ Node nodeB = nativeNode(contexts.get(1));
+ RaftNode owner = Mockito.spy(context.node());
+ AtomicBoolean interruptJoin = new AtomicBoolean(true);
+ AtomicReference<InterruptedException> joinFailure = new AtomicReference<>();
+ // Inject an interrupted wait at Node.join(); retry delegates to the same native node.
+ // Setting the flag alone is insufficient: native component joins may consume it.
+ Node joining = (Node) Proxy.newProxyInstance(Node.class.getClassLoader(),
+ new Class<?>[]{Node.class}, (proxy, method, args) -> {
+ if ("join".equals(method.getName()) && interruptJoin.get()) {
+ Thread.currentThread().interrupt();
+ try {
+ new CountDownLatch(1).await();
+ } catch (InterruptedException interrupted) {
+ joinFailure.set(interrupted);
+ throw interrupted;
+ }
+ }
+ try {
+ return method.invoke(nodeA, args);
+ } catch (InvocationTargetException failure) {
+ throw failure.getCause();
+ }
+ });
+ Whitebox.setInternalState(owner, "node", joining);
+ // Keep the baseline group-service join on the identical native join boundary too.
+ Object group = Whitebox.getInternalState(owner, "raftGroupService");
+ Whitebox.setInternalState(group, "node", joining);
+ Whitebox.setInternalState(context, "raftNode", owner);
+ Whitebox.setInternalState(manager, "metaManager", Mockito.mock(MetaManager.class));
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Thread.currentThread().setName("grpc-interrupted-raft-join");
+ HugeException failure = Assert.assertThrows(HugeException.class,
+ () -> manager.dropGraph("DEFAULT", a.name(), false));
+ Assert.assertNotNull(joinFailure.get());
+ Assert.assertSame(joinFailure.get(), HugeException.rootCause(failure));
+ Assert.assertTrue(Thread.currentThread().isInterrupted());
+ Assert.assertTrue(Thread.interrupted());
+ Assert.assertTrue(a.closed());
+ Assert.assertSame(a, manager.graph(a.spaceGraphName()));
+ Map<?, ?> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertSame(a, factory.get(a.spaceGraphName()));
+ Assert.assertTrue(nativeRpcListening(shared.port()));
+ Assert.assertTrue(NodeManager.getInstance().serverExists(address));
+ Assert.assertFalse(context.snapshotExecutor().isShutdown());
+ Assert.assertFalse(context.backendExecutor().isShutdown());
+ assertRaftLeader(client, connect, b.spaceGraphName());
+ Mockito.verify(owner).shutdown();
+ interruptJoin.set(false);
+ manager.dropGraph("DEFAULT", a.name(), false);
+ Assert.assertEquals(State.STATE_SHUTDOWN, nodeA.getNodeState());
+ assertRaftExecutorsTerminated(context);
+ Assert.assertFalse(manager.graphs().contains(a.spaceGraphName()));
+ Mockito.verify(owner, Mockito.times(2)).shutdown();
+ a.close();
+ context.close();
+ Mockito.verify(owner, Mockito.times(2)).shutdown();
+ Assert.assertEquals(State.STATE_LEADER, nodeB.getNodeState());
+ assertRaftLeader(client, connect, b.spaceGraphName());
+ client.shutdown();
+ client = null;
+ manager.close();
+ awaitNativeRpcClosed(shared.port());
+ Assert.assertFalse(NodeManager.getInstance().serverExists(address));
+ } finally {
+ Thread.interrupted();
+ Thread.currentThread().setName(threadName);
+ if (client != null) {
+ client.shutdown();
+ }
+ manager.close();
+ for (RaftContext context : contexts) {
+ stopRaftFixtureExecutors(context);
+ }
+ for (HugeGraph graph : graphs) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ private GraphManager raftOwnerManager(String name, int port) throws Exception {
+ OptionSpace.register("rocksdb", RocksDBOptions.instance());
+ BackendProviderFactory.register("rocksdb", RocksDBStoreProvider.class.getName());
+ GraphManager manager = this.dropManager(name);
+ RpcServer rpc = Whitebox.getInternalState(manager, "rpcServer");
+ ServerConfig config = Whitebox.getInternalState(rpc, "serverConfig");
+ config.setPort(port);
+ return manager;
+ }
+
+ private HugeGraph attachRaftGraph(GraphManager manager, String name,
+ com.alipay.remoting.rpc.RpcServer shared,
+ List<RaftContext> contexts) throws Exception {
+ String suffix = name + "_" + contexts.size();
+ String data = this.temporary.newFolder(suffix + "_data").getPath();
+ String wal = this.temporary.newFolder(suffix + "_wal").getPath();
+ HugeGraph graph = manager.createGraphLocal(name, "gremlin.graph=" + HugeFactory.class.getName() +
+ "\nbackend=rocksdb\nserializer=binary\nstore=" + name +
+ "\nrocksdb.data_path=" + data + "\nrocksdb.wal_path=" + wal + "\n");
+ HugeConfig config = graph.configuration();
+ config.setProperty("raft.group_peers", shared.ip() + ":" + shared.port());
+ config.setProperty(CoreOptions.RAFT_PATH.name(), this.temporary.newFolder(suffix + "_raft").getPath());
+ config.setProperty(CoreOptions.RAFT_ELECTION_TIMEOUT.name(), 300);
+ config.setProperty(CoreOptions.RAFT_SNAPSHOT_INTERVAL.name(), 0);
+ config.setProperty(CoreOptions.RAFT_RPC_THREADS.name(), 2);
+ config.setProperty(CoreOptions.RAFT_QUEUE_SIZE.name(), 2048);
+ config.setProperty(CoreOptions.RAFT_SAFE_READ.name(), false);
+ HugeGraphParams params = Whitebox.getInternalState(graph, "params");
+ RaftBackendStoreProvider raft = new RaftBackendStoreProvider(params, graph.storeProvider());
+ raft.loadSchemaStore(config);
+ raft.loadGraphStore(config);
+ raft.loadSystemStore(config);
+ contexts.add(Whitebox.getInternalState(raft, "context"));
+ Whitebox.setInternalState(graph, "storeProvider", raft);
+ return graph;
+ }
+
+ private static void startSharedRaftPair(List<RaftContext> contexts,
+ com.alipay.remoting.rpc.RpcServer shared) throws Exception {
+ contexts.get(0).initRaftNode(shared);
+ RaftContext second = contexts.get(1);
+ // Production second-context init currently repeats processor registration and fails.
+ // Bypass only that independent startup defect to exercise real shared-node close ownership.
+ Whitebox.setInternalState(second, "raftRpcServer", new BoltRpcServer(shared));
+ Whitebox.setInternalState(second, "sharedRpcServer", true);
+ Whitebox.setInternalState(second, "endpoint", new PeerId(shared.ip(), shared.port()));
+ Whitebox.setInternalState(second, "raftNode", new RaftNode(second));
+ awaitRaftLeader(nativeNode(contexts.get(0)));
+ awaitRaftLeader(nativeNode(second));
+ }
+
+ private static Node nativeNode(RaftContext context) {
+ return Whitebox.invoke(RaftNode.class, "node", context.node());
+ }
+
+ private static void awaitRaftLeader(Node node) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (node.getNodeState() != State.STATE_LEADER && System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ }
+ Assert.assertEquals(State.STATE_LEADER, node.getNodeState());
+ }
+
+ private static void assertRaftLeader(CliClientServiceImpl client, Endpoint endpoint,
+ String graph) throws Exception {
+ com.google.protobuf.Message response = client.getLeader(endpoint,
+ CliRequests.GetLeaderRequest.newBuilder().setGroupId(graph).build(), null)
+ .get(5, TimeUnit.SECONDS);
+ Assert.assertTrue(response instanceof CliRequests.GetLeaderResponse);
+ Assert.assertFalse(((CliRequests.GetLeaderResponse) response).getLeaderId().isEmpty());
+ }
+
+ @Test
+ public void testFatalCloseErrorEscapesPeerRemovalBatch() throws Exception {
+ GraphManager manager = this.dropManager("fatal_batch");
+ HugeGraph first = null;
+ HugeGraph next = null;
+ BackendStoreProvider provider = null;
+ try {
+ first = manager.createGraphLocal("fatal_first", sharedMemoryConfig("fatal_first"));
+ next = manager.createGraphLocal("fatal_next", sharedMemoryConfig("fatal_next"));
+ Map<String, HugeGraph> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ factory.put(first.spaceGraphName(), first);
+ factory.put(next.spaceGraphName(), next);
+ provider = Mockito.spy(first.storeProvider());
+ AssertionError fatal = new AssertionError("fatal provider close");
+ Mockito.doThrow(fatal).when(provider).close();
+ Whitebox.setInternalState(first, "storeProvider", provider);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Mockito.when(meta.extractGraphsFromResponse("fatal")).thenReturn(
+ List.of(first.spaceGraphName(), next.spaceGraphName()));
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ java.lang.reflect.Method method = GraphManager.class.getDeclaredMethod("graphRemoveHandler", Object.class);
+ method.setAccessible(true);
+ AssertionError caught = Assert.assertThrows(AssertionError.class, () -> {
+ try {
+ method.invoke(manager, "fatal");
+ } catch (InvocationTargetException failure) {
+ // Whitebox.invoke wraps Error, so preserve the actual handler result here.
+ throw failure.getCause();
+ }
+ });
+ Assert.assertSame(fatal, caught);
+ Assert.assertSame(first, manager.graph(first.spaceGraphName()));
+ Assert.assertSame(next, manager.graph(next.spaceGraphName()));
+ Assert.assertSame(first, factory.get(first.spaceGraphName()));
+ Assert.assertFalse(next.closed());
+ } finally {
+ if (provider != null) {
+ Mockito.doCallRealMethod().when(provider).close();
+ }
+ manager.close();
+ if (first != null) {
+ HugeFactory.remove(first);
+ }
+ if (next != null) {
+ HugeFactory.remove(next);
+ }
+ }
+ }
+
+ @Test
+ public void testEarlierRuntimeFailureDoesNotHideFatalCloseError() throws Exception {
+ this.checkFatalClosePriority(false);
+ }
+
+ @Test
+ public void testRepeatedFatalCloseFailureKeepsIdentity() throws Exception {
+ this.checkFatalClosePriority(true);
+ }
+
+ private void checkFatalClosePriority(boolean same) throws Exception {
+ String name = same ? "fatal_repeated" : "fatal_priority";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph first = null;
+ HugeGraph next = null;
+ BackendStoreProvider provider = null;
+ org.apache.hugegraph.auth.AuthManager auth = null;
+ try {
+ first = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ next = manager.createGraphLocal(name + "_next", sharedMemoryConfig(name + "_next"));
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", first);
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", next);
+ AssertionError fatal = new AssertionError("fatal native release");
+ RuntimeException earlier = new IllegalStateException("earlier close failed");
+ auth = Mockito.spy(first.authManager());
+ Mockito.doThrow(same ? fatal : earlier).when(auth).close();
+ first.switchAuthManager(auth);
+ provider = Mockito.spy(first.storeProvider());
+ Mockito.doThrow(fatal).when(provider).close();
+ Whitebox.setInternalState(first, "storeProvider", provider);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Mockito.when(meta.extractGraphsFromResponse("fatal-priority")).thenReturn(
+ List.of(first.spaceGraphName(), next.spaceGraphName()));
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ spaces.put("DEFAULT", space);
+ java.lang.reflect.Method method = GraphManager.class.getDeclaredMethod("graphRemoveHandler", Object.class);
+ method.setAccessible(true);
+ AssertionError caught = Assert.assertThrows(AssertionError.class, () -> {
+ try {
+ method.invoke(manager, "fatal-priority");
+ } catch (InvocationTargetException wrapped) {
+ throw wrapped.getCause();
+ }
+ });
+ Assert.assertSame(fatal, caught);
+ Assert.assertEquals(same ? 0 : 1, fatal.getSuppressed().length);
+ if (!same) {
+ Assert.assertSame(earlier, fatal.getSuppressed()[0]);
+ }
+ Assert.assertSame(first, manager.graph(first.spaceGraphName()));
+ Assert.assertSame(next, manager.graph(next.spaceGraphName()));
+ Assert.assertFalse(next.closed());
+ } finally {
+ if (auth != null) {
+ Mockito.doCallRealMethod().when(auth).close();
+ }
+ if (provider != null) {
+ Mockito.doCallRealMethod().when(provider).close();
+ }
+ manager.close();
+ if (first != null) {
+ HugeFactory.remove(first);
+ }
+ if (next != null) {
+ HugeFactory.remove(next);
+ }
+ }
+ }
+
+ @Test
+ public void testClosedUnclearedOwnerRejectsRestClearRetry() throws Exception {
+ String name = "closed_uncleared";
+ GraphManager manager = this.dropManager(name);
+ HugeGraph graph = null;
+ BackendStoreProvider provider = null;
+ try {
+ OptionSpace.register("rocksdb", RocksDBOptions.instance());
+ BackendProviderFactory.register("rocksdb", RocksDBStoreProvider.class.getName());
+ String config = "gremlin.graph=" + HugeFactory.class.getName() +
+ "\nbackend=rocksdb\nserializer=binary\nstore=" + name +
+ "\nrocksdb.data_path=" + this.temporary.newFolder("data").getPath() +
+ "\nrocksdb.wal_path=" + this.temporary.newFolder("wal").getPath() + "\n";
+ graph = manager.createGraphLocal(name, config);
+ graph.schema().vertexLabel("kept").useCustomizeStringId().create();
+ graph.addVertex(org.apache.tinkerpop.gremlin.structure.T.label, "kept",
+ org.apache.tinkerpop.gremlin.structure.T.id, "uncleared");
+ graph.tx().commit();
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ provider = Mockito.spy(graph.storeProvider());
+ IllegalStateException first = new IllegalStateException("provider release failed");
+ Mockito.doThrow(first).when(provider).close();
+ Whitebox.setInternalState(graph, "storeProvider", provider);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ HugeException failedClose = Assert.assertThrows(HugeException.class,
+ () -> manager.dropGraph("DEFAULT", name, false));
+ Assert.assertSame(first, failedClose.getCause());
+ Assert.assertTrue(graph.closed());
+ Map<String, HugeGraph> certificates = Whitebox.getInternalState(manager, "clearedGraphs");
+ Assert.assertFalse(certificates.containsKey(graph.spaceGraphName()));
+ Mockito.doCallRealMethod().when(provider).close();
+
+ HugeException refused = Assert.assertThrows(HugeException.class, () ->
+ new org.apache.hugegraph.api.profile.GraphsAPI().drop(
+ manager, "DEFAULT", name, "I'm sure to drop the graph"));
+ Assert.assertContains("without a successful backend clear", refused.getMessage());
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ Mockito.verifyNoInteractions(meta);
+ } finally {
+ if (provider != null) {
+ Mockito.doCallRealMethod().when(provider).close();
+ }
+ manager.close();
+ if (graph != null) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ @Test
+ public void testClearDeleteRetriesAfterForeignTransactionCloses() throws Exception {
+ this.checkClearedOwnerRetry(0);
+ }
+
+ @Test
+ public void testClearedOwnerPeerRetryDoesNotClearReplacement() throws Exception {
+ this.checkClearedOwnerRetry(1);
+ }
+
+ @Test
+ public void testManagerStopKeepsUnfinishedClearProgress() throws Exception {
+ this.checkClearedOwnerRetry(2);
+ }
+
+ private void checkClearedOwnerRetry(int retryMode) throws Exception {
+ String name = "clear_idle_retry_" + retryMode;
+ GraphManager manager = this.dropManager(name);
+ HugeGraph graph = null;
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ CountDownLatch opened = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger clears = new AtomicInteger();
+ try {
+ graph = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ HugeGraph owner = graph;
+ Map<String, HugeGraph> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ factory.put(owner.spaceGraphName(), owner);
+ HugeGraph proxy = (HugeGraph) Proxy.newProxyInstance(HugeGraph.class.getClassLoader(),
+ new Class<?>[]{HugeGraph.class}, (delegate, method, args) -> {
+ if ("clearBackend".equals(method.getName())) {
+ clears.incrementAndGet();
+ }
+ try {
+ return method.invoke(owner, args);
+ } catch (InvocationTargetException failure) {
+ throw failure.getCause();
+ }
+ });
+ Map<String, HugeGraph> graphs = Whitebox.getInternalState(manager, "graphs");
+ graphs.put(graph.spaceGraphName(), proxy);
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ spaces.put("DEFAULT", space);
+ Future<?> transaction = worker.submit(() -> {
+ try {
+ owner.tx().open();
+ opened.countDown();
+ E.checkState(release.await(15, TimeUnit.SECONDS), "Foreign transaction release timed out");
+ } catch (InterruptedException failure) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(failure);
+ } finally {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", owner);
+ }
+ });
+ Assert.assertTrue(opened.await(5, TimeUnit.SECONDS));
+ HugeException refusal = Assert.assertThrows(HugeException.class,
+ () -> manager.dropGraph("DEFAULT", name, true));
+ Assert.assertTrue(refusal.getCause() instanceof IllegalStateException);
+ Assert.assertFalse(graph.closed());
+ Assert.assertEquals(1, clears.get());
+ Map<String, HugeGraph> progress = Whitebox.getInternalState(manager, "clearedGraphs");
+ Assert.assertSame(proxy, progress.get(graph.spaceGraphName()));
+ Assert.assertSame(proxy, manager.graph(graph.spaceGraphName()));
+ if (retryMode == 1) {
+ Assert.assertThrows(HugeException.class, () -> manager.dropGraph("DEFAULT", name, false));
+ Assert.assertSame(proxy, progress.get(graph.spaceGraphName()));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ RpcServer server = Whitebox.getInternalState(manager, "rpcServer");
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 0);
+ } else if (retryMode == 2) {
+ manager.close();
+ Assert.assertSame(proxy, progress.get(graph.spaceGraphName()));
+ }
+ release.countDown();
+ transaction.get(5, TimeUnit.SECONDS);
+ if (retryMode == 2) {
+ manager.close();
+ } else {
+ // This is the same public path used by REST graph DELETE and by PD peer removal.
+ manager.dropGraph("DEFAULT", name, retryMode == 0);
+ Assert.assertFalse(manager.graphs().contains(graph.spaceGraphName()));
+ Assert.assertFalse(factory.containsKey(graph.spaceGraphName()));
+ }
+ Assert.assertTrue(graph.closed());
+ Assert.assertFalse(progress.containsKey(graph.spaceGraphName()));
+ Assert.assertEquals(1, clears.get());
+ Mockito.verify(meta).removeGraphConfig("DEFAULT", name);
+ Mockito.verify(meta).notifyGraphRemove("DEFAULT", name);
+ if (retryMode == 1) {
+ HugeGraph replacement = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ factory.put(replacement.spaceGraphName(), replacement);
+ BackendStoreProvider replacementProvider = Mockito.spy(replacement.storeProvider());
+ Whitebox.setInternalState(replacement, "storeProvider", replacementProvider);
+ // Even a stale certificate for the same name must not certify this new identity.
+ progress.put(replacement.spaceGraphName(), proxy);
+ manager.dropGraph("DEFAULT", name, true);
+ Mockito.verify(replacementProvider).clear();
+ Assert.assertTrue(replacement.closed());
+ Assert.assertFalse(progress.containsKey(replacement.spaceGraphName()));
+ Mockito.verify(meta, Mockito.times(2)).removeGraphConfig("DEFAULT", name);
+ Mockito.verify(meta, Mockito.times(2)).notifyGraphRemove("DEFAULT", name);
+ }
+ } finally {
+ release.countDown();
+ worker.shutdownNow();
+ Assert.assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS));
+ manager.close();
+ if (graph != null) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ @Test
+ public void testRaftContextClosesOwnedExecutorsAfterWork() throws Exception {
+ this.checkRaftExecutorDrain(0);
+ }
+
+ @Test
+ public void testRaftExecutorTimeoutRetainsOwnerAndQueuedWork() throws Exception {
+ this.checkRaftExecutorDrain(1);
+ }
+
+ @Test
+ public void testInterruptedRaftExecutorDrainRetainsOwner() throws Exception {
+ this.checkRaftExecutorDrain(2);
+ }
+
+ private void checkRaftExecutorDrain(int failureMode) throws Exception {
+ String name = "raft_executor_drain_" + failureMode;
+ GraphManager manager = this.dropManager(name);
+ HugeGraph graph = null;
+ RaftContext context = null;
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicBoolean interruptedWorker = new AtomicBoolean();
+ AtomicBoolean interruptAwait = new AtomicBoolean(true);
+ try {
+ graph = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ Map<String, HugeGraph> factory = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ factory.put(graph.spaceGraphName(), graph);
+ com.alipay.sofa.jraft.rpc.RpcServer rpc = startNativeRaftRpc();
+ graph.configuration().setProperty("raft.group_peers", "127.0.0.1:" + rpc.boundPort());
+ graph.configuration().setProperty(CoreOptions.RAFT_SAFE_READ.name(), true);
+ graph.configuration().setProperty(CoreOptions.RAFT_BACKEND_THREADS.name(), 1);
+ graph.configuration().setProperty(CoreOptions.TASK_WAIT_TIMEOUT.name(), failureMode == 1 ? 0L : 1L);
+ RaftBackendStoreProvider raft = new RaftBackendStoreProvider(
+ Whitebox.getInternalState(graph, "params"), graph.storeProvider());
+ context = Whitebox.getInternalState(raft, "context");
+ Whitebox.setInternalState(context, "raftRpcServer", rpc);
+ Whitebox.setInternalState(context, "endpoint", new PeerId("127.0.0.1", rpc.boundPort()));
+ Whitebox.setInternalState(graph, "storeProvider", raft);
+ for (ExecutorService executor : new ExecutorService[]{context.readIndexExecutor(),
+ context.snapshotExecutor(),
+ context.backendExecutor()}) {
+ executor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ }
+ Future<?> running = context.backendExecutor().submit(() -> {
+ started.countDown();
+ try {
+ E.checkState(release.await(10, TimeUnit.SECONDS), "Backend executor release timed out");
+ } catch (InterruptedException interrupted) {
+ interruptedWorker.set(true);
+ Thread.currentThread().interrupt();
+ }
+ });
+ Assert.assertTrue(started.await(5, TimeUnit.SECONDS));
+ AtomicInteger completed = new AtomicInteger();
+ Future<?> queued = context.backendExecutor().submit(completed::incrementAndGet);
+ Whitebox.setInternalState(manager, "metaManager", Mockito.mock(MetaManager.class));
+ Whitebox.setInternalState(manager, "PDExist", true);
+ Map<String, GraphSpace> spaces = Whitebox.getInternalState(manager, "graphSpaces");
+ GraphSpace space = new GraphSpace("DEFAULT");
+ space.graphNumberUsed(5);
+ spaces.put("DEFAULT", space);
+ if (failureMode == 0) {
+ release.countDown();
+ manager.dropGraph("DEFAULT", name, false);
+ } else {
+ if (failureMode == 2) {
+ // Begin the interrupt only at executor await, after graph's caller-TX cleanup.
+ ExecutorService backend = context.backendExecutor();
+ ExecutorService awaiting = (ExecutorService) Proxy.newProxyInstance(
+ ExecutorService.class.getClassLoader(), new Class<?>[]{ExecutorService.class},
+ (proxy, method, args) -> {
+ if ("awaitTermination".equals(method.getName()) && interruptAwait.get()) {
+ Thread.currentThread().interrupt();
+ }
+ try {
+ return method.invoke(backend, args);
+ } catch (InvocationTargetException failure) {
+ throw failure.getCause();
+ }
+ });
+ Whitebox.setInternalState(context, "backendExecutor", awaiting);
+ }
+ HugeException failure = Assert.assertThrows(HugeException.class,
+ () -> manager.dropGraph("DEFAULT", name, false));
+ if (failureMode == 2) {
+ Assert.assertTrue(HugeException.rootCause(failure) instanceof InterruptedException);
+ Assert.assertTrue(Thread.interrupted());
+ interruptAwait.set(false);
+ }
+ Assert.assertTrue(graph.closed());
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ Assert.assertSame(graph, factory.get(graph.spaceGraphName()));
+ Assert.assertTrue(context.backendExecutor().isShutdown());
+ Assert.assertFalse(context.backendExecutor().isTerminated());
+ Assert.assertFalse(queued.isDone());
+ Assert.assertFalse(interruptedWorker.get());
+ release.countDown();
+ running.get(5, TimeUnit.SECONDS);
+ queued.get(5, TimeUnit.SECONDS);
+ graph.configuration().setProperty(CoreOptions.TASK_WAIT_TIMEOUT.name(), 1L);
+ manager.dropGraph("DEFAULT", name, false);
+ }
+ running.get(5, TimeUnit.SECONDS);
+ queued.get(5, TimeUnit.SECONDS);
+ Assert.assertEquals(1, completed.get());
+ Assert.assertFalse(interruptedWorker.get());
+ assertRaftExecutorsTerminated(context);
+ Assert.assertFalse(manager.graphs().contains(graph.spaceGraphName()));
+ awaitNativeRpcClosed(rpc.boundPort());
+ } finally {
+ Thread.interrupted();
+ release.countDown();
+ stopRaftFixtureExecutors(context);
+ manager.close();
+ if (graph != null) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ private static void assertRaftExecutorsTerminated(RaftContext context) {
+ for (ExecutorService executor : new ExecutorService[]{context.readIndexExecutor(),
+ context.snapshotExecutor(),
+ context.backendExecutor()}) {
+ if (executor != null) {
+ Assert.assertTrue(executor.isShutdown());
+ Assert.assertTrue(executor.isTerminated());
+ }
+ }
+ }
+
+ private static com.alipay.sofa.jraft.rpc.RpcServer startNativeRaftRpc() throws Exception {
+ int port;
+ try (ServerSocket reservation = new ServerSocket(0)) {
+ port = reservation.getLocalPort();
+ }
+ return RaftRpcServerFactory.createAndStartRaftRpcServer(new Endpoint("127.0.0.1", port));
+ }
+
+ private static boolean nativeRpcListening(int port) {
+ try (Socket socket = new Socket()) {
+ socket.connect(new InetSocketAddress("127.0.0.1", port), 200);
+ return true;
+ } catch (Exception refused) {
+ return false;
+ }
+ }
+
+ private static void awaitNativeRpcClosed(int port) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (nativeRpcListening(port) && System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ }
+ Assert.assertFalse(nativeRpcListening(port));
+ }
+
+ private static void stopRaftFixtureExecutors(RaftContext context) throws Exception {
+ if (context == null) {
+ return;
+ }
+ for (ExecutorService pool : new ExecutorService[]{context.readIndexExecutor(), context.snapshotExecutor(),
+ context.backendExecutor()}) {
+ if (pool != null) {
+ pool.shutdownNow();
+ Assert.assertTrue(pool.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+ }
+
+ @Test
+ public void testMetadataDeleteFailureRestoresRetainedOwnerRpc() throws Exception {
+ this.checkMetadataDeleteFailure(false, false);
+ }
+
+ @Test
+ public void testMetadataDeleteFailureDoesNotSkipPeerRemove() throws Exception {
+ this.checkMetadataDeleteFailure(false, false, true);
+ }
+
+ @Test
+ public void testRemoveNotificationFailureRestoresRetainedOwnerRpc() throws Exception {
+ this.checkMetadataDeleteFailure(true, false);
+ }
+
+ @Test
+ public void testMetadataDeleteFailurePreservesRpcRestoreFailure() throws Exception {
+ this.checkMetadataDeleteFailure(false, true);
+ }
+
+ private void checkMetadataDeleteFailure(boolean notify, boolean failRestore) throws Exception {
+ this.checkMetadataDeleteFailure(notify, failRestore, false);
+ }
+
+ private void checkMetadataDeleteFailure(boolean notify, boolean failRestore, boolean peerRemove) throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ String name = notify ? "rpc_notify_fail" : failRestore ? "rpc_restore_fail" : "rpc_metadata_fail";
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder(name).getPath());
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false);
+ properties.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "127.0.0.1");
+ properties.setProperty(RpcOptions.RPC_SERVER_PORT.name(), 0);
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "127.0.0.1:1");
+ EventHub hub = new EventHub(name);
+ GraphManager manager = new GraphManager(new HugeConfig(properties), hub);
+ HugeGraph graph = null;
+ String threadName = Thread.currentThread().getName();
+ try {
+ graph = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ Whitebox.invoke(GraphManager.class, "startRpcServer", manager);
+ RpcServer original = Whitebox.getInternalState(manager, "rpcServer");
+ RpcServer server = Mockito.spy(original);
+ Whitebox.setInternalState(manager, "rpcServer", server);
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ assertRpcResources(server, client, 2);
+ IllegalStateException failure = new IllegalStateException("metadata fault before clear");
+ IllegalStateException restoration = new IllegalStateException("RPC restoration export failed");
+ if (failRestore) {
+ Mockito.doThrow(restoration).when(server).exportAll();
+ }
+ MetaManager meta = Mockito.mock(MetaManager.class);
+ if (notify) {
+ Mockito.doThrow(failure).doNothing().when(meta).notifyGraphRemove("DEFAULT", name);
+ } else {
+ Mockito.doThrow(failure).doNothing().when(meta).removeGraphConfig("DEFAULT", name);
+ }
+ Whitebox.setInternalState(manager, "metaManager", meta);
+ Whitebox.setInternalState(manager, "PDExist", true);
+ AtomicReference<Throwable> rejected = new AtomicReference<>();
+ Assert.assertThrows(HugeException.class, () -> manager.dropGraph("DEFAULT", name, true), rejected::set);
+ Mockito.verify(meta).removeGraphConfig("DEFAULT", name);
+ if (notify) {
+ Mockito.verify(meta).notifyGraphRemove("DEFAULT", name);
+ } else {
+ Mockito.verify(meta, Mockito.never()).notifyGraphRemove("DEFAULT", name);
+ }
+ Assert.assertFalse(graph.closed());
+ Assert.assertSame(graph, manager.graph(graph.spaceGraphName()));
+ assertRpcResources(server, client, failRestore ? 0 : 2);
+ Assert.assertSame(failure, rejected.get().getCause());
+ Mockito.verify(server).exportAll();
+ Assert.assertEquals(failRestore ? 1 : 0, failure.getSuppressed().length);
+ if (failRestore) {
+ Assert.assertSame(restoration, failure.getSuppressed()[0]);
+ }
+ Thread.currentThread().setName("grpc-rpc-metadata-retry");
+ if (peerRemove) {
+ // Another server deleted the config after this local metadata attempt failed.
+ Mockito.when(meta.extractGraphsFromResponse(Mockito.any()))
+ .thenReturn(java.util.Collections.singletonList(graph.spaceGraphName()));
+ Whitebox.invoke(GraphManager.class, new Class<?>[]{Object.class},
+ "graphRemoveHandler", manager, "peer-remove");
+ } else {
+ // Retry the fenced deletion; successful clear must not re-export the removed owner.
+ if (notify) {
+ java.util.Set<?> removing = Whitebox.getInternalState(manager, "removingGraphs");
+ Assert.assertTrue(removing.contains(graph.spaceGraphName()));
+ }
+ manager.dropGraph("DEFAULT", name, true);
+ }
+ Assert.assertTrue(graph.closed());
+ Assert.assertFalse(manager.graphs().contains(graph.spaceGraphName()));
+ assertRpcResources(server, client, 0);
+ Mockito.verify(server).exportAll();
+ } finally {
+ Thread.currentThread().setName(threadName);
+ manager.close();
+ if (graph != null) {
+ HugeFactory.remove(graph);
+ }
+ }
+ }
+
+ @Test
+ public void testLocalGraphGremlinPublicationWaitsForRpc() throws Exception {
+ this.checkLocalGraphGremlinPublication(false);
+ }
+
+ @Test
+ public void testLocalGraphRpcFailureNeverPublishesToGremlin() throws Exception {
+ this.checkLocalGraphGremlinPublication(true);
+ }
+
+ private void checkLocalGraphGremlinPublication(boolean failExport) throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ String name = failExport ? "rpc_export_failure" : "rpc_export_success";
+ String graphName = "DEFAULT-" + name;
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder(name).getPath());
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false);
+ properties.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "127.0.0.1");
+ properties.setProperty(RpcOptions.RPC_SERVER_PORT.name(), 0);
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "127.0.0.1:1");
+ EventHub hub = new EventHub(name);
+ Settings settings = new Settings();
+ settings.gremlinPool = 1;
+ settings.threadPoolBoss = 1;
+ settings.threadPoolWorker = 1;
+ // Production constructs Gremlin before REST/GraphManager, registering its listener first.
+ ContextGremlinServer gremlin = new ContextGremlinServer(settings, hub);
+ GraphManager manager = new GraphManager(new HugeConfig(properties), hub);
+ RpcServer original = Whitebox.getInternalState(manager, "rpcServer");
+ RpcServer server = Mockito.spy(original);
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ Whitebox.setInternalState(manager, "rpcServer", server);
+ Whitebox.setInternalState(manager, "rpcInitialized", true);
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ IllegalStateException failure = new IllegalStateException("RPC export rejected");
+ Mockito.doAnswer(invocation -> {
+ entered.countDown();
+ E.checkState(release.await(10, TimeUnit.SECONDS), "RPC export barrier timed out");
+ if (failExport) {
+ throw failure;
+ }
+ return invocation.callRealMethod();
+ }).when(server).exportAll();
+ ExecutorService initializer = Executors.newSingleThreadExecutor();
+ AtomicReference<HugeGraph> candidate = new AtomicReference<>();
+ org.apache.tinkerpop.gremlin.server.GraphManager bindings =
+ gremlin.getServerGremlinExecutor().getGraphManager();
+ try {
+ Future<HugeGraph> publication = initializer.submit(() -> {
+ try {
+ HugeGraph graph = manager.createGraphLocal(name, sharedMemoryConfig(name));
+ candidate.set(graph);
+ return graph;
+ } finally {
+ Map<String, HugeGraph> owners = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ HugeGraph graph = owners.get(graphName);
+ if (graph != null) {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ }
+ }
+ });
+ Assert.assertTrue(entered.await(5, TimeUnit.SECONDS));
+ Assert.assertNull(bindings.getGraph(graphName));
+ Assert.assertNull(bindings.getTraversalSource("__g_" + graphName));
+ Assert.assertFalse(bindings.getAsBindings().containsKey(graphName));
+ Assert.assertFalse(manager.graphs().contains(graphName));
+ assertRpcResources(server, client, 2);
+ release.countDown();
+ if (failExport) {
+ Assert.assertThrows(ExecutionException.class, () -> publication.get(5, TimeUnit.SECONDS),
+ caught -> Assert.assertSame(failure, caught.getCause()));
+ Assert.assertNull(bindings.getGraph(graphName));
+ Assert.assertNull(bindings.getTraversalSource("__g_" + graphName));
+ Assert.assertFalse(manager.graphs().contains(graphName));
+ Map<?, ?> owners = Whitebox.getInternalState(HugeFactory.class, "GRAPHS");
+ Assert.assertFalse(owners.containsKey(graphName));
+ assertRpcResources(server, client, 0);
+ } else {
+ HugeGraph graph = publication.get(5, TimeUnit.SECONDS);
+ Assert.assertSame(graph, bindings.getGraph(graphName));
+ Assert.assertNotNull(bindings.getTraversalSource("__g_" + graphName));
+ Assert.assertSame(graph, manager.graph(graphName));
+ assertRpcResources(server, client, 2);
+ }
+ } finally {
+ release.countDown();
+ initializer.shutdown();
+ Assert.assertTrue(initializer.awaitTermination(10, TimeUnit.SECONDS));
+ manager.close();
+ gremlin.stop().get(30, TimeUnit.SECONDS);
+ if (candidate.get() != null) {
+ HugeFactory.remove(candidate.get());
+ }
+ HugeGraphAuthProxy.resetContext();
+ }
+ }
+
+ @Test
+ public void testLateGraphRpcPublicationRollbackAndReplacement() throws Exception {
+ OptionSpace.register("core", CoreOptions.instance());
+ OptionSpace.register("rpc", RpcOptions.instance());
+ OptionSpace.register("server", ServerOptions.instance());
+ PropertiesConfiguration properties = new PropertiesConfiguration();
+ properties.setProperty(ServerOptions.GRAPHS.name(), this.temporary.newFolder("graphs").getPath());
+ properties.setProperty(ServerOptions.USE_PD.name(), false);
+ properties.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false);
+ properties.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true);
+ properties.setProperty(RpcOptions.RPC_SERVER_HOST.name(), "127.0.0.1");
+ properties.setProperty(RpcOptions.RPC_SERVER_PORT.name(), 0);
+ try (ServerSocket reservation = new ServerSocket(0)) {
+ properties.setProperty(RpcOptions.RPC_REMOTE_URL.name(), "127.0.0.1:" + reservation.getLocalPort());
+ }
+ EventHub hub = new EventHub("rpc-publication-test");
+ GraphManager manager = new GraphManager(new HugeConfig(properties), hub);
+ BarrierRpcServer server = new BarrierRpcServer(new HugeConfig(properties));
+ Whitebox.setInternalState(manager, "rpcServer", server);
+ RpcClientProvider client = Whitebox.getInternalState(manager, "rpcClient");
+ ExecutorService initializer = Executors.newFixedThreadPool(2);
+ HugeGraph startup = null;
+ HugeGraph replacement = null;
+ try {
+ startup = manager.createGraphLocal("rpc_startup", sharedMemoryConfig("rpc_startup"));
+ Assert.assertEquals(0, server.config().configs().size());
+ HugeGraph startupOwner = startup;
+ AtomicReference<Thread> initThread = new AtomicReference<>();
+ Future<?> initialization = initializer.submit(() -> {
+ initThread.set(Thread.currentThread());
+ try {
+ Whitebox.invoke(manager.getClass(), "startRpcServer", manager);
+ } finally {
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", startupOwner);
+ }
+ });
+ Assert.assertTrue(server.entered.await(5, TimeUnit.SECONDS));
+ Assert.assertEquals(2, server.config().configs().size());
+ AtomicReference<Thread> publisher = new AtomicReference<>();
+ Future<HugeGraph> publication = initializer.submit(() -> {
+ publisher.set(Thread.currentThread());
+ HugeGraph graph = manager.createGraphLocal("rpc_late", sharedMemoryConfig("rpc_late"));
+ Whitebox.invoke(StandardHugeGraph.class, "closeCurrentThreadTransaction", graph);
+ return graph;
+ });
+ awaitManagerGate(publisher, initThread.get());
+ Assert.assertFalse(manager.graphs().contains("DEFAULT-rpc_late"));
+ assertRpcResources(server, client, 2);
+ server.release.countDown();
+ initialization.get(5, TimeUnit.SECONDS);
+ HugeGraph original = publication.get(5, TimeUnit.SECONDS);
+ Assert.assertEquals(2, server.exports.get());
+ assertRpcResources(server, client, 4);
+
+ EventListener failure = event -> {
+ throw new IllegalStateException("reject candidate after RPC registration");
+ };
+ hub.listen(Events.GRAPH_CREATE, failure);
+ try {
+ Assert.assertThrows(HugeException.class, () -> manager.createGraphLocal(
+ "rpc_rejected", sharedMemoryConfig("rpc_rejected")));
+ } finally {
+ hub.unlisten(Events.GRAPH_CREATE, failure);
+ }
+ Assert.assertFalse(manager.graphs().contains("DEFAULT-rpc_rejected"));
+ assertRpcResources(server, client, 4);
+
+ manager.dropGraphLocal("rpc_late");
+ assertRpcResources(server, client, 2);
+ replacement = manager.createGraphLocal("rpc_late", sharedMemoryConfig("rpc_late"));
+ hub.notifySync(Events.GRAPH_DROP, original);
+ Assert.assertSame(replacement, manager.graph("DEFAULT-rpc_late"));
+ assertRpcResources(server, client, 4);
+ manager.close();
+ List<?> bootstraps = Whitebox.getInternalState(client.config(), "bootstraps");
+ Assert.assertEquals(0, server.config().configs().size());
+ Assert.assertEquals(0, bootstraps.size());
+ } finally {
+ server.release.countDown();
+ initializer.shutdownNow();
+ initializer.awaitTermination(5, TimeUnit.SECONDS);
+ manager.close();
+ if (startup != null) {
+ HugeFactory.remove(startup);
+ }
+ if (replacement != null) {
+ HugeFactory.remove(replacement);
+ }
+ }
+ }
+
+ private static void awaitManagerGate(AtomicReference<Thread> publisher, Thread initializer) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (System.nanoTime() < deadline) {
+ Thread thread = publisher.get();
+ ThreadInfo info = thread == null ? null : ManagementFactory.getThreadMXBean().getThreadInfo(thread.getId());
+ if (info != null && info.getThreadState() == Thread.State.BLOCKED &&
+ info.getLockOwnerId() == initializer.getId()) {
+ Assert.assertContains(GraphManager.class.getName(), info.getLockName());
+ return;
+ }
+ Thread.sleep(5);
+ }
+ Assert.fail("Graph publisher did not block on the startup manager gate");
+ }
+
+ private static class BarrierRpcServer extends RpcServer {
+
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final AtomicInteger exports = new AtomicInteger();
+
+ BarrierRpcServer(HugeConfig config) {
+ super(config);
+ }
+
+ @Override
+ public void exportAll() {
+ if (this.exports.incrementAndGet() == 1) {
+ this.entered.countDown();
+ try {
+ E.checkState(this.release.await(5, TimeUnit.SECONDS), "RPC export barrier timed out");
+ } catch (InterruptedException failure) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(failure);
+ }
+ }
+ super.exportAll();
+ }
+ }
+
+ private static void assertRpcResources(RpcServer server, RpcClientProvider client, int count) {
+ Assert.assertEquals(count, server.config().configs().size());
+ Map<?, ?> configs = Whitebox.getInternalState(client.config(), "configs");
+ List<?> bootstraps = Whitebox.getInternalState(client.config(), "bootstraps");
+ Assert.assertEquals(count, configs.size());
+ Assert.assertEquals(count, bootstraps.size());
+ }
+
+ private static String sharedMemoryConfig(String name) {
+ return "gremlin.graph=" + SharedMemoryGraph.class.getName() + "\nbackend=memory\nstore=" + name + "\n";
+ }
+
+ // Exercises cache RPC registration, not shared-backend DML or consistency.
+ public static class SharedMemoryGraph extends StandardHugeGraph {
+
+ public SharedMemoryGraph(HugeConfig config) {
+ super(config);
+ }
+
+ public static SharedMemoryGraph open(Configuration config) {
+ return new SharedMemoryGraph(new HugeConfig(config));
+ }
+
+ @Override
+ public BackendFeatures backendStoreFeatures() {
+ BackendFeatures features = super.backendStoreFeatures();
+ return (BackendFeatures) Proxy.newProxyInstance(BackendFeatures.class.getClassLoader(),
+ new Class<?>[]{BackendFeatures.class}, (proxy, method, args) ->
+ "supportsSharedStorage".equals(method.getName()) ? true : method.invoke(features, args));
+ }
+ }
+
@Test
public void testAttachExplicitLocalResourceLimits() {
PropertiesConfiguration properties = new PropertiesConfiguration();