fix: sync hstore schema cache clears

Register a JVM-wide schema cache clear listener for CachedSchemaTransactionV2 so HStore server nodes can invalidate local schema caches when another node changes schema.

Clear V2 schema-id/name caches and the attached array cache by graph name, and publish schema cache clear events after schema add, update, and remove paths.

Add unit coverage for graph-scoped V2 schema cache clearing.

Fixes #2617

Author: dpol1

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java

@@ -19,8 +19,10 @@
 
 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.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 
 import org.apache.hugegraph.HugeGraphParams;
@@ -43,6 +45,13 @@
 
 public class CachedSchemaTransactionV2 extends SchemaTransactionV2 {
 
+    private static final String ID_CACHE_PREFIX = "schema-id";
+    private static final String NAME_CACHE_PREFIX = "schema-name";
+
+    // MetaDriver doesn't expose unlisten, register the PD listener once.
+    private static final AtomicBoolean metaEventListenerRegistered =
+            new AtomicBoolean(false);
+
     private final Cache<Id, Object> idCache;
     private final Cache<Id, Object> nameCache;
 
@@ -58,8 +67,8 @@ public CachedSchemaTransactionV2(MetaDriver metaDriver,
 
         final long capacity = graphParams.configuration()
                                          .get(CoreOptions.SCHEMA_CACHE_CAPACITY);
-        this.idCache = this.cache("schema-id", capacity);
-        this.nameCache = this.cache("schema-name", capacity);
+        this.idCache = this.cache(ID_CACHE_PREFIX, capacity);
+        this.nameCache = this.cache(NAME_CACHE_PREFIX, capacity);
 
         SchemaCaches<SchemaElement> attachment = this.idCache.attachment();
         if (attachment == null) {
@@ -86,11 +95,36 @@ public void close() {
     }
 
     private Cache<Id, Object> cache(String prefix, long capacity) {
-        final String name = prefix + "-" + this.graph().spaceGraphName();
+        final String name = cacheName(prefix, this.graph().spaceGraphName());
         // NOTE: must disable schema cache-expire due to getAllSchema()
         return CacheManager.instance().cache(name, capacity);
     }
 
+    private static String cacheName(String prefix, String spaceGraphName) {
+        return prefix + "-" + spaceGraphName;
+    }
+
+    private static void clearSchemaCache(String spaceGraphName) {
+        Map<String, Cache<Id, Object>> caches = CacheManager.instance().caches();
+
+        Cache<Id, Object> idCache = caches.get(cacheName(ID_CACHE_PREFIX,
+                                                         spaceGraphName));
+        if (idCache != null) {
+            idCache.clear();
+
+            SchemaCaches<?> arrayCaches = idCache.attachment();
+            if (arrayCaches != null) {
+                arrayCaches.clear();
+            }
+        }
+
+        Cache<Id, Object> nameCache = caches.get(cacheName(NAME_CACHE_PREFIX,
+                                                           spaceGraphName));
+        if (nameCache != null) {
+            nameCache.clear();
+        }
+    }
+
     private void listenChanges() {
         // Listen store event: "store.init", "store.clear", ...
         Set<String> storeEvents = ImmutableSet.of(Events.STORE_INIT,
@@ -142,15 +176,44 @@ private void listenChanges() {
             return false;
         };
         EventHub schemaEventHub = this.graphParams().schemaEventHub();
-        if (!schemaEventHub.containsListener(Events.CACHE)) {
-            schemaEventHub.listen(Events.CACHE, this.cacheEventListener);
+        schemaEventHub.listen(Events.CACHE, this.cacheEventListener);
+
+        listenSchemaCacheClear();
+    }
+
+    private static void listenSchemaCacheClear() {
+        if (!metaEventListenerRegistered.compareAndSet(false, true)) {
+            return;
+        }
+
+        try {
+            MetaDriver metaDriver = MetaManager.instance().metaDriver();
+            MetaManager.instance().listenSchemaCacheClear(response -> {
+                List<String> graphNames =
+                        metaDriver.extractValuesFromResponse(response);
+                if (graphNames == null) {
+                    return;
+                }
+                for (String graphName : graphNames) {
+                    LOG.debug("Graph {} clear schema cache on meta event",
+                              graphName);
+                    clearSchemaCache(graphName);
+                }
+            });
+        } catch (RuntimeException e) {
+            metaEventListenerRegistered.set(false);
+            throw e;
         }
     }
 
     public void clearCache(boolean notify) {
         this.idCache.clear();
         this.nameCache.clear();
         this.arrayCaches.clear();
+
+        if (notify) {
+            this.notifySchemaCacheClear();
+        }
     }
 
     private void resetCachedAllIfReachedCapacity() {
@@ -202,6 +265,7 @@ protected void updateSchema(SchemaElement schema,
         super.updateSchema(schema, updateCallback);
 
         this.updateCache(schema);
+        this.notifySchemaCacheClear();
     }
 
     @Override
@@ -210,11 +274,7 @@ protected void addSchema(SchemaElement schema) {
 
         this.updateCache(schema);
 
-        if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
-            MetaManager.instance()
-                       .notifySchemaCacheClear(this.graph().graphSpace(),
-                                               this.graph().name());
-        }
+        this.notifySchemaCacheClear();
     }
 
     private void updateCache(SchemaElement schema) {
@@ -238,6 +298,10 @@ public void removeSchema(SchemaElement schema) {
 
         this.invalidateCache(schema.type(), schema.id());
 
+        this.notifySchemaCacheClear();
+    }
+
+    private void notifySchemaCacheClear() {
         if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
             MetaManager.instance()
                        .notifySchemaCacheClear(this.graph().graphSpace(),

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/cache/CachedSchemaTransactionTest.java

@@ -20,7 +20,11 @@
 import org.apache.hugegraph.HugeFactory;
 import org.apache.hugegraph.HugeGraph;
 import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.backend.cache.Cache;
+import org.apache.hugegraph.backend.cache.CacheManager;
 import org.apache.hugegraph.backend.cache.CachedSchemaTransaction;
+import org.apache.hugegraph.backend.cache.CachedSchemaTransactionV2;
+import org.apache.hugegraph.backend.id.Id;
 import org.apache.hugegraph.backend.id.IdGenerator;
 import org.apache.hugegraph.testutil.Assert;
 import org.apache.hugegraph.testutil.Whitebox;
@@ -165,6 +169,41 @@ public void testGetSchema() throws Exception {
                             cache.getPropertyKey(IdGenerator.of(1)).name());
     }
 
+    @Test
+    public void testClearV2SchemaCacheByGraphName() {
+        String graphName = "DEFAULT-unit-test-v2";
+        String otherGraphName = "DEFAULT-other-v2";
+
+        Cache<Id, Object> idCache = CacheManager.instance()
+                                                .cache("schema-id-" +
+                                                       graphName, 10L);
+        Cache<Id, Object> nameCache = CacheManager.instance()
+                                                  .cache("schema-name-" +
+                                                         graphName, 10L);
+        Cache<Id, Object> otherIdCache = CacheManager.instance()
+                                                     .cache("schema-id-" +
+                                                            otherGraphName,
+                                                            10L);
+
+        idCache.update(IdGenerator.of(1), "fake-pk-by-id");
+        nameCache.update(IdGenerator.of("fake-pk"), "fake-pk-by-name");
+        otherIdCache.update(IdGenerator.of(2), "other-pk-by-id");
+
+        Assert.assertEquals(1L, idCache.size());
+        Assert.assertEquals(1L, nameCache.size());
+        Assert.assertEquals(1L, otherIdCache.size());
+
+        Whitebox.invokeStatic(CachedSchemaTransactionV2.class,
+                              new Class<?>[]{String.class},
+                              "clearSchemaCache", graphName);
+
+        Assert.assertEquals(0L, idCache.size());
+        Assert.assertEquals(0L, nameCache.size());
+        Assert.assertEquals(1L, otherIdCache.size());
+
+        otherIdCache.clear();
+    }
+
     @Test
     public void testResetCachedAllIfReachedCapacity() throws Exception {
         CachedSchemaTransaction cache = this.cache();