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 c9a646dd8 fix(server): wait for GRAPH_CREATE event when creating graph
on PD path (#3138)
c9a646dd8 is described below
commit c9a646dd820dd331eab97c73c913f9f4a81f4030
Author: KAI <[email protected]>
AuthorDate: Wed Aug 12 09:05:14 2026 +0530
fix(server): wait for GRAPH_CREATE event when creating graph on PD path
(#3138)
This is Phase 1 of #3137. It closes the local race on the Server that
handles graph creation; cluster-wide readiness remains a follow-up.
In distributed mode (PD + HStore), graph creation previously returned HTTP
200 after scheduling `GRAPH_CREATE`, but before the embedded Gremlin Server had
necessarily registered the graph and its `TraversalSource`.
---------
Co-authored-by: imbajin <[email protected]>
---
.../java/org/apache/hugegraph/event/EventHub.java | 86 +++++++++++++++-----
.../apache/hugegraph/unit/event/EventHubTest.java | 40 +++++++++
.../hugegraph/auth/ContextGremlinServer.java | 18 ++--
.../org/apache/hugegraph/core/GraphManager.java | 95 ++++++++++++++++------
4 files changed, 190 insertions(+), 49 deletions(-)
diff --git
a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java
b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java
index fbdf460a9..0787b783d 100644
---
a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java
+++
b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java
@@ -39,6 +39,29 @@ import com.google.common.collect.ImmutableList;
public class EventHub {
+ public static final class NotifyResult {
+
+ private final int attempted;
+ private final int succeeded;
+
+ private NotifyResult(int attempted, int succeeded) {
+ this.attempted = attempted;
+ this.succeeded = succeeded;
+ }
+
+ public int attempted() {
+ return this.attempted;
+ }
+
+ public int succeeded() {
+ return this.succeeded;
+ }
+
+ public boolean success() {
+ return this.attempted == this.succeeded;
+ }
+ }
+
private static final Logger LOG = Log.logger(EventHub.class);
public static final String EVENT_WORKER = "event-worker-%d";
@@ -167,10 +190,29 @@ public class EventHub {
return this.notify(event, null, args);
}
+ /**
+ * Notify all registered listeners in the current thread.
+ */
+ public NotifyResult notifySync(String event, @Nullable Object... args) {
+ ExtendableIterator<EventListener> all = this.eventListeners(event);
+ return this.notifyListeners(all, null, new Event(this, event, args));
+ }
+
private Future<Integer> notify(String event,
EventListener ignoredListener,
@Nullable Object... args) {
- @SuppressWarnings("resource")
+ ExtendableIterator<EventListener> all = this.eventListeners(event);
+ if (!all.hasNext()) {
+ return CompletableFuture.completedFuture(0);
+ }
+ Event ev = new Event(this, event, args);
+ return executor().submit(() -> {
+ return this.notifyListeners(all, ignoredListener, ev).succeeded();
+ });
+ }
+
+ @SuppressWarnings("resource")
+ private ExtendableIterator<EventListener> eventListeners(String event) {
ExtendableIterator<EventListener> all = new ExtendableIterator<>();
List<EventListener> ls = this.listeners.get(event);
@@ -181,31 +223,33 @@ public class EventHub {
if (lsAny != null && !lsAny.isEmpty()) {
all.extend(lsAny.iterator());
}
+ return all;
+ }
+ private NotifyResult notifyListeners(ExtendableIterator<EventListener> all,
+ EventListener ignoredListener,
+ Event event) {
if (!all.hasNext()) {
- return CompletableFuture.completedFuture(0);
+ return new NotifyResult(0, 0);
}
- Event ev = new Event(this, event, args);
-
- // The submit will catch params: `all`(Listeners) and `ev`(Event)
- return executor().submit(() -> {
- int count = 0;
- // Notify all listeners, and ignore the results
- while (all.hasNext()) {
- EventListener listener = all.next();
- if (listener == ignoredListener) {
- continue;
- }
- try {
- listener.event(ev);
- count++;
- } catch (Throwable e) {
- LOG.warn("Failed to handle event: {}", ev, e);
- }
+ int attempted = 0;
+ int succeeded = 0;
+ // Notify all listeners, and ignore the results
+ while (all.hasNext()) {
+ EventListener listener = all.next();
+ if (listener == ignoredListener) {
+ continue;
}
- return count;
- });
+ attempted++;
+ try {
+ listener.event(event);
+ succeeded++;
+ } catch (Throwable e) {
+ LOG.warn("Failed to handle event: {}", event, e);
+ }
+ }
+ return new NotifyResult(attempted, succeeded);
}
public Object call(String event, @Nullable Object... args) {
diff --git
a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java
b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java
index 69472bc8e..43e4721e6 100644
---
a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java
+++
b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java
@@ -31,6 +31,7 @@ import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.unit.BaseUnitTest;
import org.apache.hugegraph.event.Event;
import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventHub.NotifyResult;
import org.apache.hugegraph.event.EventListener;
public class EventHubTest extends BaseUnitTest {
@@ -388,6 +389,45 @@ public class EventHubTest extends BaseUnitTest {
Assert.assertEquals(1, count.get());
}
+ @Test
+ public void testEventNotifySync() {
+ final String notify = "event-notify-sync";
+ AtomicInteger count = new AtomicInteger();
+
+ this.eventHub.listen(notify, event -> {
+ count.incrementAndGet();
+ return null;
+ });
+ this.eventHub.listen(notify, event -> {
+ throw new RuntimeException("fake exception");
+ });
+ this.eventHub.listen(EventHub.ANY_EVENT, event -> {
+ count.incrementAndGet();
+ return null;
+ });
+
+ NotifyResult result = this.eventHub.notifySync(notify);
+ Assert.assertEquals(3, result.attempted());
+ Assert.assertEquals(2, result.succeeded());
+ Assert.assertFalse(result.success());
+ Assert.assertEquals(2, count.get());
+ }
+
+ @Test
+ public void testEventNotifySyncUsesSingleListenerSnapshot() {
+ final String notify = "event-notify-sync-snapshot";
+ this.eventHub.listen(notify, event -> {
+ this.eventHub.listen(notify, ignored -> null);
+ return null;
+ });
+
+ NotifyResult result = this.eventHub.notifySync(notify);
+
+ Assert.assertEquals(1, result.attempted());
+ Assert.assertEquals(1, result.succeeded());
+ Assert.assertTrue(result.success());
+ }
+
@Test
public void testNotifyExcept() throws Exception {
final String notify = "event-notify";
diff --git
a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java
b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java
index 0f5881b1a..e85905013 100644
---
a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java
+++
b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java
@@ -75,7 +75,7 @@ public class ContextGremlinServer extends GremlinServer {
LOG.debug("GremlinServer accepts event '{}'", event.name());
event.checkArgs(HugeGraph.class);
HugeGraph graph = (HugeGraph) event.args()[0];
- this.removeGraph(graph.spaceGraphName());
+ this.removeGraph(graph);
return null;
});
}
@@ -123,7 +123,7 @@ public class ContextGremlinServer extends GremlinServer {
}
}
- private void injectGraph(HugeGraph graph) {
+ private synchronized void injectGraph(HugeGraph graph) {
String name = graph.spaceGraphName();
GraphManager manager = this.getServerGremlinExecutor()
.getGraphManager();
@@ -140,17 +140,25 @@ public class ContextGremlinServer extends GremlinServer {
"put", name, graph);
}
- private void removeGraph(String name) {
+ private synchronized void removeGraph(HugeGraph graph) {
+ String name = graph.spaceGraphName();
GraphManager manager = this.getServerGremlinExecutor()
.getGraphManager();
GremlinExecutor executor = this.getServerGremlinExecutor()
.getGremlinExecutor();
try {
- manager.removeGraph(name);
- manager.removeTraversalSource(G_PREFIX + name);
+ if (manager.getGraph(name) != graph) {
+ return;
+ }
+ if (manager.getTraversalSource(G_PREFIX + name) != null) {
+ manager.removeTraversalSource(G_PREFIX + name);
+ }
Whitebox.invoke(executor, "globalBindings",
new Class<?>[]{Object.class},
"remove", name);
+ if (manager.getGraph(name) != null) {
+ manager.removeGraph(name);
+ }
} catch (Exception e) {
throw new HugeException("Failed to remove graph '%s' from " +
"gremlin server context", e, name);
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 40d65c928..96717e724 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,7 +34,6 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
@@ -66,6 +65,7 @@ import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.ServerOptions;
import org.apache.hugegraph.config.TypedOption;
import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventHub.NotifyResult;
import org.apache.hugegraph.exception.ExistedException;
import org.apache.hugegraph.exception.NotFoundException;
import org.apache.hugegraph.exception.NotSupportException;
@@ -1228,30 +1228,38 @@ 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
+ this.notifyEvent(Events.GRAPH_CREATE, graph);
} catch (Throwable e) {
LOG.error("Failed to create graph '{}' due to: {}",
name, e.getMessage(), e);
if (graph != null) {
- this.dropGraphLocal(graph);
+ this.graphs.remove(graph.spaceGraphName(), graph);
+ try {
+ this.dropGraphLocal(graph);
+ } finally {
+ // The create event may have partially registered the graph
+ this.notifyEventLenient(Events.GRAPH_DROP, graph);
+ }
}
throw e;
}
- // Let gremlin server and rest server add graph to context
- this.notifyAndWaitEvent(Events.GRAPH_CREATE, graph);
-
return graph;
}
private void dropGraphLocal(HugeGraph graph) {
- // Clear data and config files
- graph.drop();
-
- /*
- * Will fill graph instance into HugeFactory.graphs after
- * GraphFactory.open() succeed, remove it when the graph drops
- */
- HugeFactory.remove(graph);
+ try {
+ // Clear data and config files
+ graph.drop();
+ } finally {
+ /*
+ * Will fill graph instance into HugeFactory.graphs after
+ * GraphFactory.open() succeed, remove it when the graph drops
+ */
+ HugeFactory.remove(graph);
+ }
}
public HugeGraph createGraph(String graphSpace, String name, String
creator,
@@ -1377,19 +1385,38 @@ 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
+ * 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.notifyEvent(Events.GRAPH_CREATE, graph);
+ } catch (Throwable e) {
+ this.notifyEventLenient(Events.GRAPH_DROP, graph);
+ this.graphs.remove(graphName, graph);
+ 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());
@@ -1789,7 +1816,7 @@ public final class GraphManager {
LOG.debug("RestServer accepts event '{}'", event.name());
event.checkArgs(HugeGraph.class);
HugeGraph graph = (HugeGraph) event.args()[0];
- this.graphs.remove(graph.spaceGraphName());
+ this.graphs.remove(graph.spaceGraphName(), graph);
return null;
});
}
@@ -1806,12 +1833,34 @@ public final class GraphManager {
this.metaManager.listenGraphClear(ConsumerWrapper.wrap(this::graphClearHandler));
}
- private void notifyAndWaitEvent(String event, HugeGraph graph) {
- Future<?> future = this.eventHub.notify(event, graph);
+ /**
+ * Notify the listeners of `event` synchronously, failing if any listener
+ * did not complete successfully.
+ * <p>
+ * EventHub swallows every throwable raised by a listener and reports the
+ * attempted and successful listeners from the same snapshot.
+ */
+ private void notifyEvent(String event, HugeGraph graph) {
+ String graphName = graph.spaceGraphName();
+ NotifyResult result = this.eventHub.notifySync(event, graph);
+
+ if (!result.success()) {
+ throw new HugeException("Only %s of %s listeners handled event " +
+ "'%s' of graph '%s' successfully",
+ result.succeeded(), result.attempted(),
+ event, graphName);
+ }
+ }
+
+ /**
+ * Notify listeners synchronously, but keep listener failures non-fatal.
+ * Used by the drop and rollback paths, where cleanup must be best-effort.
+ */
+ private void notifyEventLenient(String event, HugeGraph graph) {
try {
- future.get();
+ this.eventHub.notifySync(event, graph);
} catch (Throwable e) {
- LOG.warn("Error when waiting for event execution: {}", event, e);
+ LOG.warn("Error when notifying event: {}", event, e);
}
}
@@ -2034,7 +2083,7 @@ public final class GraphManager {
this.dropGraphLocal(graph);
// Let gremlin server and rest server context remove graph
- this.notifyAndWaitEvent(Events.GRAPH_DROP, graph);
+ this.notifyEventLenient(Events.GRAPH_DROP, graph);
}
public void dropGraph(String graphSpace, String name, boolean clear) {
