Skip to content

fix(server): align cache event actions in legacy EventHub path #3017

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
imbajin merged 4 commits into from
May 15, 2026
Merged

fix(server): align cache event actions in legacy EventHub path #3017

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
aa4f37b
fix(server): align legacy cache event actions
dpol1 May 8, 2026
4369458
fix(test): stabilize cache expiration unit test
dpol1 May 8, 2026
80bd29c
fix(server): ref-count cache event listeners
dpol1 May 9, 2026
5f64206
fix(cache): address reviewer comments on cache listener cleanup
dpol1 May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 25 additions & 1 deletion hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/event/EventHub.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,27 @@ public int unlisten(String event, EventListener listener) {
return count;
}

/**
* Notify all registered listeners for {@code event} EXCEPT
* {@code ignoredListener}. ANY_EVENT listeners are notified unless they
* are the ignored one.
*
* @return a Future<Integer> resolving to the count of listeners actually
* invoked (the ignored listener is NOT counted)
*/
public Future<Integer> notifyExcept(String event,
EventListener ignoredListener,
@Nullable Object... args) {
return this.notify(event, ignoredListener, args);
}

public Future<Integer> notify(String event, @Nullable Object... args) {
return this.notify(event, null, args);
}

private Future<Integer> notify(String event,
EventListener ignoredListener,
@Nullable Object... args) {
@SuppressWarnings("resource")
ExtendableIterator<EventListener> all = new ExtendableIterator<>();

Expand All @@ -173,8 +193,12 @@ public Future<Integer> notify(String event, @Nullable Object... args) {
int count = 0;
// Notify all listeners, and ignore the results
while (all.hasNext()) {
EventListener listener = all.next();
if (listener == ignoredListener) {
continue;
}
try {
all.next().event(ev);
listener.event(ev);
count++;
} catch (Throwable e) {
LOG.warn("Failed to handle event: {}", ev, e);
Expand Down
37 changes: 37 additions & 0 deletions ...-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/event/EventHubTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -388,6 +388,43 @@ public void testEventNotifyWithArg2() {
Assert.assertEquals(1, count.get());
}

@Test
public void testNotifyExcept() throws Exception {
final String notify = "event-notify";
AtomicInteger listenerACount = new AtomicInteger();
AtomicInteger listenerBCount = new AtomicInteger();
AtomicInteger listenerCCount = new AtomicInteger();

EventListener listenerA = event -> {
event.checkArgs(String.class);
Assert.assertEquals("fake-arg", event.args()[0]);
listenerACount.incrementAndGet();
return true;
};
EventListener listenerB = event -> {
listenerBCount.incrementAndGet();
return true;
};
EventListener listenerC = event -> {
event.checkArgs(String.class);
Assert.assertEquals("fake-arg", event.args()[0]);
listenerCCount.incrementAndGet();
return true;
};

this.eventHub.listen(notify, listenerA);
this.eventHub.listen(notify, listenerB);
this.eventHub.listen(EventHub.ANY_EVENT, listenerC);

Assert.assertEquals(2, (int) this.eventHub
.notifyExcept(notify, listenerB,
"fake-arg")
.get());
Assert.assertEquals(1, listenerACount.get());
Assert.assertEquals(0, listenerBCount.get());
Assert.assertEquals(1, listenerCCount.get());
}

@Test
public void testEventNotifyWithMultiThreads() throws InterruptedException {
final String notify = "event-notify";
Expand Down
13 changes: 8 additions & 5 deletions hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1394,7 +1394,7 @@ public AbstractCacheNotifier(EventHub hub, CacheNotifier proxy) {
"Expect event action argument");
String action = (String) args[0];
LOG.debug("Event action: {}", action);
if (Cache.ACTION_INVALIDED.equals(action)) {
if (Cache.ACTION_INVALID.equals(action)) {
event.checkArgs(String.class, HugeType.class, Object.class);
HugeType type = (HugeType) args[1];
Object ids = args[2];
Expand All @@ -1410,7 +1410,7 @@ public AbstractCacheNotifier(EventHub hub, CacheNotifier proxy) {
E.checkArgument(false, "Unexpected argument: %s", ids);
}
return true;
} else if (Cache.ACTION_CLEARED.equals(action)) {
} else if (Cache.ACTION_CLEAR.equals(action)) {
event.checkArgs(String.class, HugeType.class);
HugeType type = (HugeType) args[1];
LOG.debug("Calling proxy.clear with type: {}", type);
Expand All @@ -1435,17 +1435,20 @@ public void close() {

@Override
public void invalid(HugeType type, Id id) {
this.hub.notify(Events.CACHE, Cache.ACTION_INVALID, type, id);
this.hub.notifyExcept(Events.CACHE, this.cacheEventListener,
Cache.ACTION_INVALID, type, id);
}

@Override
public void invalid2(HugeType type, Object[] ids) {
this.hub.notify(Events.CACHE, Cache.ACTION_INVALID, type, ids);
this.hub.notifyExcept(Events.CACHE, this.cacheEventListener,
Cache.ACTION_INVALID, type, ids);
}

@Override
public void clear(HugeType type) {
this.hub.notify(Events.CACHE, Cache.ACTION_CLEAR, type);
this.hub.notifyExcept(Events.CACHE, this.cacheEventListener,
Cache.ACTION_CLEAR, type);
}

@Override
Expand Down
2 changes: 0 additions & 2 deletions hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/Cache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,6 @@ public interface Cache<K, V> {

String ACTION_INVALID = "invalid";
String ACTION_CLEAR = "clear";
String ACTION_INVALIDED = "invalided";
String ACTION_CLEARED = "cleared";

V get(K id);

Expand Down
37 changes: 26 additions & 11 deletions ...gegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedGraphTransaction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ public final class CachedGraphTransaction extends GraphTransaction {

private EventListener storeEventListener;
private EventListener cacheEventListener;
private boolean registeredCacheEventListener;

public CachedGraphTransaction(HugeGraphParams graph, BackendStore store) {
super(graph, store);
Expand Down Expand Up @@ -138,7 +139,7 @@ private void listenChanges() {
}

// Listen cache event: "cache"(invalid cache item)
this.cacheEventListener = event -> {
EventListener listener = event -> {
LOG.debug("Graph {} received graph cache event: {}",
this.graph(), event);
Object[] args = event.args();
Expand Down Expand Up @@ -184,17 +185,29 @@ private void listenChanges() {
}
return false;
};
if (graphCacheListenStatus.putIfAbsent(this.params().spaceGraphName(), true) == null) {
EventHub graphEventHub = this.params().graphEventHub();
EventHub graphEventHub = this.params().graphEventHub();
EventListener previous =
graphCacheEventListeners.putIfAbsent(
this.params().spaceGraphName(), listener);
if (previous == null) {
this.cacheEventListener = listener;
this.registeredCacheEventListener = true;
graphEventHub.listen(Events.CACHE, this.cacheEventListener);
} else {
this.cacheEventListener = previous;
this.registeredCacheEventListener = false;
}
}

private void unlistenChanges() {
String graphName = this.params().spaceGraphName();
if (graphCacheListenStatus.remove(graphName) != null) {
if (this.registeredCacheEventListener) {
EventHub graphEventHub = this.params().graphEventHub();
graphEventHub.unlisten(Events.CACHE, this.cacheEventListener);
if (graphCacheEventListeners.remove(graphName,
this.cacheEventListener)) {
graphEventHub.unlisten(Events.CACHE, this.cacheEventListener);
}
this.registeredCacheEventListener = false;
Comment thread
dpol1 marked this conversation as resolved.
Outdated
Comment thread
dpol1 marked this conversation as resolved.
Outdated
}
if (storeEventListenStatus.remove(graphName) != null) {
this.store().provider().unlisten(this.storeEventListener);
Expand All @@ -203,12 +216,14 @@ private void unlistenChanges() {

private void notifyChanges(String action, HugeType type, Id[] ids) {
EventHub graphEventHub = this.params().graphEventHub();
graphEventHub.notify(Events.CACHE, action, type, ids);
graphEventHub.notifyExcept(Events.CACHE, this.cacheEventListener,
action, type, ids);
}

private void notifyChanges(String action, HugeType type) {
EventHub graphEventHub = this.params().graphEventHub();
graphEventHub.notify(Events.CACHE, action, type);
graphEventHub.notifyExcept(Events.CACHE, this.cacheEventListener,
action, type);
}

public void clearCache(HugeType type, boolean notify) {
Expand All @@ -220,7 +235,7 @@ public void clearCache(HugeType type, boolean notify) {
}

if (notify) {
this.notifyChanges(Cache.ACTION_CLEARED, null);
this.notifyChanges(Cache.ACTION_CLEAR, null);
}
}

Expand Down Expand Up @@ -397,7 +412,7 @@ protected void commitMutation2Backend(BackendMutation... mutations) {
this.verticesCache.invalidate(vertex.id());
}
if (vertexOffset > 0) {
this.notifyChanges(Cache.ACTION_INVALIDED,
this.notifyChanges(Cache.ACTION_INVALID,
HugeType.VERTEX, vertexIds);
}
}
Expand All @@ -411,7 +426,7 @@ protected void commitMutation2Backend(BackendMutation... mutations) {
if (invalidEdgesCache && this.enableCacheEdge()) {
// TODO: Use a more precise strategy to update the edge cache
this.edgesCache.clear();
this.notifyChanges(Cache.ACTION_CLEARED, HugeType.EDGE);
this.notifyChanges(Cache.ACTION_CLEAR, HugeType.EDGE);
}
}
}
Expand All @@ -425,7 +440,7 @@ public void removeIndex(IndexLabel indexLabel) {
if (indexLabel.baseType() == HugeType.EDGE_LABEL) {
// TODO: Use a more precise strategy to update the edge cache
this.edgesCache.clear();
this.notifyChanges(Cache.ACTION_CLEARED, HugeType.EDGE);
this.notifyChanges(Cache.ACTION_CLEAR, HugeType.EDGE);
}
}
}
Expand Down
38 changes: 29 additions & 9 deletions ...egraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransaction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;

import org.apache.hugegraph.HugeGraphParams;
Expand All @@ -42,13 +43,17 @@

public final class CachedSchemaTransaction extends SchemaTransaction {

private static final ConcurrentMap<String, EventListener>
SCHEMA_CACHE_EVENT_LISTENERS = new ConcurrentHashMap<>();

private final Cache<Id, Object> idCache;
private final Cache<Id, Object> nameCache;

private final SchemaCaches<SchemaElement> arrayCaches;

private EventListener storeEventListener;
private EventListener cacheEventListener;
private boolean registeredCacheEventListener;

public CachedSchemaTransaction(HugeGraphParams graph, BackendStore store) {
super(graph, store);
Expand Down Expand Up @@ -111,7 +116,7 @@ private void listenChanges() {
this.store().provider().listen(this.storeEventListener);

// Listen cache event: "cache"(invalid cache item)
this.cacheEventListener = event -> {
EventListener listener = event -> {
LOG.debug("Graph {} received schema cache event: {}",
this.graph(), event);
Object[] args = event.args();
Expand All @@ -132,8 +137,16 @@ private void listenChanges() {
return false;
};
EventHub schemaEventHub = this.params().schemaEventHub();
if (!schemaEventHub.containsListener(Events.CACHE)) {
String graph = this.params().spaceGraphName();
EventListener previous =
SCHEMA_CACHE_EVENT_LISTENERS.putIfAbsent(graph, listener);
if (previous == null) {
this.cacheEventListener = listener;
this.registeredCacheEventListener = true;
schemaEventHub.listen(Events.CACHE, this.cacheEventListener);
} else {
this.cacheEventListener = previous;
this.registeredCacheEventListener = false;
}
}

Expand All @@ -143,17 +156,24 @@ private void unlistenChanges() {

// Unlisten cache event
EventHub schemaEventHub = this.params().schemaEventHub();
schemaEventHub.unlisten(Events.CACHE, this.cacheEventListener);
if (this.registeredCacheEventListener) {
schemaEventHub.unlisten(Events.CACHE, this.cacheEventListener);
SCHEMA_CACHE_EVENT_LISTENERS.remove(this.params().spaceGraphName(),
this.cacheEventListener);
this.registeredCacheEventListener = false;
}
Comment thread
dpol1 marked this conversation as resolved.
}

private void notifyChanges(String action, HugeType type, Id id) {
EventHub graphEventHub = this.params().schemaEventHub();
graphEventHub.notify(Events.CACHE, action, type, id);
graphEventHub.notifyExcept(Events.CACHE, this.cacheEventListener,
action, type, id);
}

private void notifyChanges(String action, HugeType type) {
EventHub graphEventHub = this.params().schemaEventHub();
graphEventHub.notify(Events.CACHE, action, type);
graphEventHub.notifyExcept(Events.CACHE, this.cacheEventListener,
action, type);
}

private void resetCachedAll(HugeType type) {
Expand All @@ -179,7 +199,7 @@ public void clearCache(boolean notify) {
this.arrayCaches.clear();

if (notify) {
this.notifyChanges(Cache.ACTION_CLEARED, null);
this.notifyChanges(Cache.ACTION_CLEAR, null);
}
}

Expand Down Expand Up @@ -221,7 +241,7 @@ protected void updateSchema(SchemaElement schema,

this.updateCache(schema);

this.notifyChanges(Cache.ACTION_INVALIDED, schema.type(), schema.id());
this.notifyChanges(Cache.ACTION_INVALID, schema.type(), schema.id());
}

@Override
Expand All @@ -230,7 +250,7 @@ protected void addSchema(SchemaElement schema) {

this.updateCache(schema);

this.notifyChanges(Cache.ACTION_INVALIDED, schema.type(), schema.id());
this.notifyChanges(Cache.ACTION_INVALID, schema.type(), schema.id());
}

@Override
Expand Down Expand Up @@ -283,7 +303,7 @@ public void removeSchema(SchemaElement schema) {

this.invalidateCache(schema.type(), schema.id());

this.notifyChanges(Cache.ACTION_INVALIDED, schema.type(), schema.id());
this.notifyChanges(Cache.ACTION_INVALID, schema.type(), schema.id());
}

@Override
Expand Down
5 changes: 3 additions & 2 deletions ...server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/GraphTransaction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
import org.apache.hugegraph.backend.store.BackendStore;
import org.apache.hugegraph.config.CoreOptions;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.event.EventListener;
import org.apache.hugegraph.exception.LimitExceedException;
import org.apache.hugegraph.exception.NotFoundException;
import org.apache.hugegraph.iterator.BatchMapperIterator;
Expand Down Expand Up @@ -140,8 +141,8 @@ public class GraphTransaction extends IndexableTransaction {

private final int verticesCapacity;
private final int edgesCapacity;
protected static final ConcurrentHashMap<String, Boolean> graphCacheListenStatus =
new ConcurrentHashMap<>();
protected static final ConcurrentHashMap<String, EventListener>
graphCacheEventListeners = new ConcurrentHashMap<>();
protected static final ConcurrentHashMap<String, Boolean> storeEventListenStatus =
new ConcurrentHashMap<>();

Expand Down
Loading
Loading