fix(server): wait for GRAPH_CREATE event when creating graph on PD path

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.
+     * <p>
+     * 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<Integer> 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<Integer> 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) {