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 f716285c67..83f907a36d 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 @@ -34,8 +34,10 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -151,6 +153,13 @@ public final class GraphManager { public static final String DELIMITER = "-"; public static final String NAMESPACE_CREATE = "namespace_create"; private static final Logger LOG = Log.logger(GraphManager.class); + /* + * The graph create/drop listeners only do in-memory registrations (put the + * graph into the rest server context and into the gremlin server bindings), + * so they finish in microseconds on a healthy server. The bound is only a + * guard against a stuck or starved event worker. + */ + private static final long EVENT_WAIT_TIMEOUT = 30L; private KvStore kvStore; private final String cluster; @@ -1193,18 +1202,20 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { // Init graph and start it graph.create(this.graphsDir, this.globalNodeRoleInfo); + + // Let gremlin server and rest server add graph to context + this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); } catch (Throwable e) { LOG.error("Failed to create graph '{}' due to: {}", name, e.getMessage(), e); if (graph != null) { + // The create event may have added the graph to the context + this.graphs.remove(graph.spaceGraphName()); this.dropGraphLocal(graph); } throw e; } - // Let gremlin server and rest server add graph to context - this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); - return graph; } @@ -1342,19 +1353,37 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, graph.updateTime(timeStamp); String graphName = spaceGraphName(graphSpace, name); + this.graphs.put(graphName, graph); + + /* + * Let gremlin server and rest server context add graph before the + * graph is published, so that a failed local binding can't leave the + * graph behind in meta for the other servers to converge on + */ + try { + this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph); + } catch (Throwable e) { + this.graphs.remove(graphName); + try { + graph.close(); + } catch (Exception e1) { + if (graph instanceof StandardHugeGraph) { + ((StandardHugeGraph) graph).clearSchedulerAndLock(); + } + } + HugeFactory.remove(graph); + throw e; + } + if (init) { this.creatingGraphs.add(graphName); this.metaManager.addGraphConfig(graphSpace, name, configs); this.metaManager.notifyGraphAdd(graphSpace, name); } - this.graphs.put(graphName, graph); if (!grpcThread) { this.metaManager.updateGraphSpaceConfig(graphSpace, gs); } - // Let gremlin server and rest server context add graph - this.eventHub.notify(Events.GRAPH_CREATE, graph); - if (init) { String schema = propConfig.getString( CoreOptions.SCHEMA_INIT_TEMPLATE.name()); @@ -1771,10 +1800,60 @@ private void listenMetaChanges() { this.metaManager.listenGraphClear(ConsumerWrapper.wrap(this::graphClearHandler)); } + /** + * Notify the listeners of `event` and wait for them to finish, failing if + * any listener did not complete successfully. + *

+ * EventHub swallows every throwable raised by a listener and resolves the + * future with the number of listeners that returned normally, so waiting + * alone doesn't prove that the graph was registered. Comparing the + * notified count with the registered listener count detects the swallowed + * failure and lets the caller fail instead of returning a graph that is + * missing from the rest/gremlin server context. + */ private void notifyAndWaitEvent(String event, HugeGraph graph) { - Future future = this.eventHub.notify(event, graph); + String graphName = graph.spaceGraphName(); + // Listeners of ANY_EVENT are notified too, so they count as expected + int expected = this.eventHub.listeners(event).size() + + this.eventHub.listeners(EventHub.ANY_EVENT).size(); + int notified; try { - future.get(); + Future future = this.eventHub.notify(event, graph); + notified = future.get(EVENT_WAIT_TIMEOUT, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new HugeException("Interrupted while waiting for event " + + "'%s' of graph '%s'", e, event, graphName); + } catch (TimeoutException e) { + throw new HugeException("Timeout(%ss) while waiting for event " + + "'%s' of graph '%s'", e, + EVENT_WAIT_TIMEOUT, event, graphName); + } catch (ExecutionException e) { + throw new HugeException("Failed to wait for event '%s' of " + + "graph '%s'", e, event, graphName); + } + + if (notified < expected) { + throw new HugeException("Only %s of %s listeners handled event " + + "'%s' of graph '%s' successfully", + notified, expected, event, graphName); + } + } + + /** + * Same bounded wait as notifyAndWaitEvent(), but a failed listener is only + * logged. Used by the drop path: the graph data is already deleted when + * the event is sent, so failing the request can't undo anything, and the + * listener state may legitimately be absent already. + */ + private void notifyAndWaitEventLenient(String event, HugeGraph graph) { + try { + Future future = this.eventHub.notify(event, graph); + future.get(EVENT_WAIT_TIMEOUT, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOG.warn("Interrupted when waiting for event execution: {}", + event, e); } catch (Throwable e) { LOG.warn("Error when waiting for event execution: {}", event, e); } @@ -1999,7 +2078,7 @@ public void dropGraphLocal(String name) { this.dropGraphLocal(graph); // Let gremlin server and rest server context remove graph - this.notifyAndWaitEvent(Events.GRAPH_DROP, graph); + this.notifyAndWaitEventLenient(Events.GRAPH_DROP, graph); } public void dropGraph(String graphSpace, String name, boolean clear) {