This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new efe98bb652 [#11232][#11233] improvement(core): Global
EntityChangeLogPoller with CatalogManager cache invalidation and automatic
retention cleanup (#11254)
efe98bb652 is described below
commit efe98bb652b1885cc9d603b5cf89dddd4cf2e071
Author: Qi Yu <[email protected]>
AuthorDate: Tue Jun 2 16:32:16 2026 +0800
[#11232][#11233] improvement(core): Global EntityChangeLogPoller with
CatalogManager cache invalidation and automatic retention cleanup (#11254)
### What changes were proposed in this pull request?
1. Extract entity change log polling into a shared
`EntityChangeLogPoller` in `core` with a listener-based dispatch
pattern, replacing the per-consumer polling in `JcasbinChangePoller`.
2. Add `CatalogChangeLogListener` that invalidates `CatalogManager`'s
local catalog cache when catalog ALTER/RENAME/DROP changes are consumed,
providing eventual consistency across HA nodes.
3. Add automatic retention cleanup for `entity_change_log` — expired
rows are pruned periodically (default: 1-hour interval, 1-day
retention), with configurable settings.
4. Write change log on all catalog updates (not just renames) so that
property-only changes also trigger cross-node cache invalidation.
### Why are the changes needed?
In a multi-node deployment, catalog changes on one server do not
invalidate the `CatalogManager` cache on peer servers until local TTL
eviction. This can leave peers using stale catalog entities or class
loaders. Additionally, `entity_change_log` grows unboundedly without
automatic cleanup.
Fix: #11232
Fix: #11233
### Does this PR introduce _any_ user-facing change?
Yes — three new configuration keys:
- `gravitino.entityChangeLog.pollIntervalSecs` (default 3)
- `gravitino.entityChangeLog.retentionSecs` (default 86400)
- `gravitino.entityChangeLog.cleanupIntervalSecs` (default 3600)
### How was this patch tested?
- `TestEntityChangeLogPoller` — unit tests for dispatch, cursor
advancement, listener failure semantics, retention pruning, and cleanup
interval gating.
-
`TestCatalogManager.testCatalogChangeLogListenerInvalidatesCatalogCache`
— verifies cache invalidation on entity change.
- `TestEntityChangeLogService` — extended to cover ALTER (non-rename)
change log writing.
- `TestJcasbinChangePoller` — updated for listener-based entity change
delivery.
---
.../fileset/TestFilesetCatalogOperations.java | 6 +
.../catalog/kafka/TestKafkaCatalogOperations.java | 6 +
.../generic/TestGenericCatalogOperations.java | 6 +
.../catalog/model/TestModelCatalogOperations.java | 6 +
.../cache/it/AbstractEntityStorageBenchmark.java | 6 +
.../main/java/org/apache/gravitino/Configs.java | 28 +++
.../java/org/apache/gravitino/GravitinoEnv.java | 2 +
.../catalog/CatalogChangeLogListener.java | 103 +++++++++
.../apache/gravitino/catalog/CatalogManager.java | 82 +++++++
.../relational/EntityChangeLogListener.java | 34 +++
.../storage/relational/EntityChangeLogPoller.java | 248 +++++++++++++++++++++
.../storage/relational/RelationalEntityStore.java | 27 ++-
.../relational/SupportsEntityChangeLog.java | 50 +++++
.../relational/service/CatalogMetaService.java | 4 +-
.../authorization/TestAccessControlManager.java | 6 +
.../gravitino/authorization/TestOwnerManager.java | 6 +
.../gravitino/catalog/TestCatalogManager.java | 183 +++++++++++++++
.../gravitino/hook/TestFilesetHookDispatcher.java | 6 +
.../gravitino/hook/TestTableHookDispatcher.java | 6 +
.../apache/gravitino/policy/TestPolicyManager.java | 6 +
.../gravitino/stats/TestStatisticManager.java | 6 +
.../storage/AbstractEntityStorageTest.java | 6 +
.../relational/TestEntityChangeLogPoller.java | 242 ++++++++++++++++++++
.../service/TestEntityChangeLogService.java | 31 ++-
.../org/apache/gravitino/tag/TestTagManager.java | 6 +
docs/gravitino-server-config.md | 3 +
.../jcasbin/JcasbinAuthorizationLookups.java | 33 +--
.../authorization/jcasbin/JcasbinAuthorizer.java | 24 +-
...hangePoller.java => JcasbinChangeListener.java} | 53 +----
.../jcasbin/TestJcasbinAuthorizationLookups.java | 40 +++-
.../jcasbin/TestJcasbinChangePoller.java | 72 ++----
31 files changed, 1201 insertions(+), 136 deletions(-)
diff --git
a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
index 492cdcc38b..77dbf1f319 100644
---
a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
+++
b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
@@ -19,6 +19,9 @@
package org.apache.gravitino.catalog.fileset;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD;
@@ -240,6 +243,9 @@ public class TestFilesetCatalogOperations {
when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
+ when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
+ when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
Mockito.when(config.get(Configs.CACHE_MAX_ENTRIES)).thenReturn(10_000);
diff --git
a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
index 0c0f650c68..16b4951236 100644
---
a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
+++
b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
@@ -20,6 +20,9 @@ package org.apache.gravitino.catalog.kafka;
import static org.apache.gravitino.Catalog.Type.MESSAGING;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD;
@@ -159,6 +162,9 @@ public class TestKafkaCatalogOperations extends
KafkaClusterEmbedded {
when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
+ when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
+ when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
Mockito.when(config.get(Configs.CACHE_MAX_ENTRIES)).thenReturn(10_000);
diff --git
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java
index 8f61c12182..fe749a18a3 100644
---
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java
+++
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java
@@ -19,6 +19,9 @@
package org.apache.gravitino.catalog.lakehouse.generic;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD;
@@ -104,6 +107,9 @@ public class TestGenericCatalogOperations {
when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
+ when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
+ when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(false);
store = EntityStoreFactory.createEntityStore(config);
diff --git
a/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java
b/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java
index 65ea324266..3237e96105 100644
---
a/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java
+++
b/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java
@@ -19,6 +19,9 @@
package org.apache.gravtitino.catalog.model;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD;
@@ -123,6 +126,9 @@ public class TestModelCatalogOperations {
when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
+ when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
+ when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
Mockito.when(config.get(Configs.CACHE_MAX_ENTRIES)).thenReturn(10_000);
diff --git
a/core/src/jmh/java/org/apache/gravitino/cache/it/AbstractEntityStorageBenchmark.java
b/core/src/jmh/java/org/apache/gravitino/cache/it/AbstractEntityStorageBenchmark.java
index d55119a152..ba733f5f16 100644
---
a/core/src/jmh/java/org/apache/gravitino/cache/it/AbstractEntityStorageBenchmark.java
+++
b/core/src/jmh/java/org/apache/gravitino/cache/it/AbstractEntityStorageBenchmark.java
@@ -20,6 +20,9 @@
package org.apache.gravitino.cache.it;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD;
@@ -290,6 +293,9 @@ public class AbstractEntityStorageBenchmark<E extends
Entity & HasIdentifier> {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS)).thenReturn(100);
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(CACHE_ENABLED);
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index 516c2c9106..6319cdd739 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -184,6 +184,34 @@ public class Configs {
.longConf()
.createWithDefault(60 * 60 * 1000L);
+ public static final long DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS = 3L;
+ public static final long DEFAULT_ENTITY_CHANGE_LOG_RETENTION_SECS = 24 * 60
* 60L;
+ public static final long DEFAULT_ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS =
60 * 60L;
+
+ public static final ConfigEntry<Long> ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS =
+ new ConfigBuilder("gravitino.entityChangeLog.pollIntervalSecs")
+ .doc("The interval in seconds for polling entity change logs")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .longConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS);
+
+ public static final ConfigEntry<Long> ENTITY_CHANGE_LOG_RETENTION_SECS =
+ new ConfigBuilder("gravitino.entityChangeLog.retentionSecs")
+ .doc("The retention time in seconds for entity change logs. Set 0 to
disable cleanup")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .longConf()
+ .checkValue(value -> value >= 0,
ConfigConstants.NON_NEGATIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_RETENTION_SECS);
+
+ public static final ConfigEntry<Long>
ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS =
+ new ConfigBuilder("gravitino.entityChangeLog.cleanupIntervalSecs")
+ .doc("The interval in seconds for pruning expired entity change
logs")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .longConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS);
+
public static final ConfigEntry<Boolean> CATALOG_LOAD_ISOLATED =
new ConfigBuilder("gravitino.catalog.classloader.isolated")
.doc("Whether to load the catalog in an isolated classloader")
diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
index 7514d2b930..df80fad2ca 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -564,6 +564,8 @@ public class GravitinoEnv {
// Create and initialize Catalog related modules, the operation chain is:
// CatalogHookDispatcher -> CatalogEventDispatcher ->
CatalogNormalizeDispatcher ->
// CatalogManager
+ // CatalogManager registers its own change-log listener with the entity
store (when the store
+ // supports it), so no poller wiring is needed here.
this.catalogManager = new CatalogManager(config, entityStore, idGenerator);
CatalogNormalizeDispatcher catalogNormalizeDispatcher =
new CatalogNormalizeDispatcher(catalogManager);
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
new file mode 100644
index 0000000000..218e2d1319
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Optional;
+import org.apache.gravitino.Entity.EntityType;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.storage.relational.EntityChangeLogListener;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Invalidates {@link CatalogManager}'s local catalog cache from {@code
entity_change_log}.
+ *
+ * <p>This listener is called <em>synchronously</em> in the poller thread.
Implementations must not
+ * block or perform expensive I/O; only fast, in-memory cache invalidations
are permitted.
+ */
+public class CatalogChangeLogListener implements EntityChangeLogListener {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CatalogChangeLogListener.class);
+
+ private final CatalogManager catalogManager;
+
+ /**
+ * Creates a listener for a catalog manager.
+ *
+ * @param catalogManager the catalog manager whose local cache should be
invalidated
+ */
+ public CatalogChangeLogListener(CatalogManager catalogManager) {
+ this.catalogManager = catalogManager;
+ }
+
+ @Override
+ public void onEntityChange(List<EntityChangeRecord> changes) {
+ for (EntityChangeRecord change : changes) {
+ try {
+ if (!isCatalogChange(change)) {
+ continue;
+ }
+
+ Optional<NameIdentifier> identOpt = catalogIdentifier(change);
+ if (identOpt.isEmpty()) {
+ continue;
+ }
+ NameIdentifier ident = identOpt.get();
+
+ if (catalogManager.consumeLocalMutation(ident)) {
+ LOG.debug("Skipping catalog cache invalidation for local mutation:
{}", ident);
+ continue;
+ }
+
+ LOG.debug("Invalidating catalog cache due to entity change log: {}",
ident);
+ catalogManager.getCatalogCache().invalidate(ident);
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Failed to process catalog change log record: fullName={},
entityType={}",
+ change.getFullName(),
+ change.getEntityType(),
+ e);
+ }
+ }
+ }
+
+ private boolean isCatalogChange(EntityChangeRecord change) {
+ if (change.getEntityType() == null) {
+ return false;
+ }
+ return
EntityType.CATALOG.name().equals(change.getEntityType().toUpperCase(Locale.ROOT));
+ }
+
+ private Optional<NameIdentifier> catalogIdentifier(EntityChangeRecord
change) {
+ if (change.getFullName() == null) {
+ LOG.warn("Invalid catalog full name in entity change log: null");
+ return Optional.empty();
+ }
+
+ String[] names = change.getFullName().split("\\.");
+ if (names.length != 2) {
+ LOG.warn("Invalid catalog full name in entity change log: {}",
change.getFullName());
+ return Optional.empty();
+ }
+ return Optional.of(NameIdentifier.of(names[0], names[1]));
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index 45d022fc71..203b8c5555 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -29,6 +29,7 @@ import static
org.apache.gravitino.metalake.MetalakeManager.checkMetalake;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Scheduler;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
@@ -51,8 +52,10 @@ import java.util.Optional;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
@@ -103,6 +106,7 @@ import org.apache.gravitino.rel.Table;
import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.ViewCatalog;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.relational.SupportsEntityChangeLog;
import org.apache.gravitino.utils.IsolatedClassLoader;
import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.gravitino.utils.PrincipalUtils;
@@ -290,8 +294,17 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
private final EntityStore store;
+ @Nullable private final CatalogChangeLogListener catalogChangeLogListener;
+
private final IdGenerator idGenerator;
private final List<Consumer<NameIdentifier>> removalListeners =
Lists.newArrayList();
+ private final ConcurrentHashMap<NameIdentifier, AtomicInteger>
localMutationCounts =
+ new ConcurrentHashMap<>();
+
+ // Set to true when a CatalogChangeLogListener is active.
markLocalMutation() is a no-op
+ // unless this flag is set, preventing unbounded growth of
localMutationCounts in deployments
+ // that do not use a relational entity store (where the poller never runs).
+ private volatile boolean trackLocalMutations = false;
/**
* Constructs a CatalogManager instance.
@@ -328,6 +341,21 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
.setNameFormat("catalog-cleaner-%d")
.build())))
.build();
+
+ // If the entity store maintains a change log, register a listener that
invalidates this
+ // manager's local catalog cache from cross-node changes, and enable
local-mutation tracking so
+ // changes made by this node are not redundantly re-invalidated.
Registration is the last step
+ // of
+ // the constructor so the listener never sees a half-initialized manager
(catalogCache and
+ // localMutationCounts are already set). CatalogChangeLogListener stays an
implementation detail
+ // of this class rather than being wired externally.
+ if (store instanceof SupportsEntityChangeLog) {
+ this.trackLocalMutations = true;
+ this.catalogChangeLogListener = new CatalogChangeLogListener(this);
+ ((SupportsEntityChangeLog)
store).registerEntityChangeLogListener(catalogChangeLogListener);
+ } else {
+ this.catalogChangeLogListener = null;
+ }
}
/**
@@ -336,6 +364,11 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
*/
@Override
public void close() {
+ if (catalogChangeLogListener != null) {
+ ((SupportsEntityChangeLog)
store).unregisterEntityChangeLogListener(catalogChangeLogListener);
+ trackLocalMutations = false;
+ localMutationCounts.clear();
+ }
catalogCache.invalidateAll();
}
@@ -354,6 +387,48 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
removalListeners.add(listener);
}
+ /**
+ * Records that this process has just mutated the given catalog locally. The
entity change log
+ * poller will see the corresponding change log row and should skip cache
invalidation for it
+ * because this process already updated the cache.
+ *
+ * @param ident the catalog identifier (pre-mutation name for renames)
+ */
+ void markLocalMutation(NameIdentifier ident) {
+ if (!trackLocalMutations) {
+ return;
+ }
+ localMutationCounts.computeIfAbsent(ident, k -> new
AtomicInteger()).incrementAndGet();
+ }
+
+ /**
+ * Attempts to consume one local mutation marker for the given identifier.
+ *
+ * <p>Thread-safety note: {@link ConcurrentHashMap#computeIfPresent}
executes the remapping
+ * function atomically under a per-bucket lock, so {@code consumed[0]} is
set and the counter is
+ * decremented as a single atomic step. {@code consumed[0]} is a
single-element array (the
+ * standard Java pattern for a mutable capture in a lambda) that is only
read by this thread after
+ * {@code computeIfPresent} returns, so no additional synchronization is
needed.
+ *
+ * @return true if a local mutation was pending and has been consumed
(caller should skip
+ * invalidation), false if the change originated from a remote node
+ */
+ boolean consumeLocalMutation(NameIdentifier ident) {
+ boolean[] consumed = {false};
+ localMutationCounts.computeIfPresent(
+ ident,
+ (k, count) -> {
+ consumed[0] = true;
+ return count.decrementAndGet() <= 0 ? null : count;
+ });
+ return consumed[0];
+ }
+
+ @VisibleForTesting
+ void setTrackLocalMutations(boolean trackLocalMutations) {
+ this.trackLocalMutations = trackLocalMutations;
+ }
+
/**
* Lists the catalogs within the specified namespace.
*
@@ -612,6 +687,7 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
return newCatalogBuilder.build();
});
+ markLocalMutation(ident);
catalogCache.invalidate(ident);
return null;
} catch (IOException e) {
@@ -652,6 +728,7 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
return newCatalogBuilder.build();
});
+ markLocalMutation(ident);
catalogCache.invalidate(ident);
return null;
} catch (IOException e) {
@@ -738,6 +815,7 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
// the old catalog identifier from the store (after the
invalidate) will get
// NoSuchCatalogException instead of stale data. Invalidating
before the update creates
// a window where the background thread repopulates the cache with
the old entity.
+ markLocalMutation(ident);
catalogCache.invalidate(ident);
// The old fileset catalog's provider is "hadoop", whereas the new
fileset catalog's
// provider is "fileset", still using "hadoop" will lead to
catalog loading issue. So
@@ -812,6 +890,9 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
// Invalidate after store.delete() to prevent a background thread
from repopulating
// the cache with stale data between invalidate and delete.
boolean deleted = store.delete(ident, EntityType.CATALOG, true);
+ if (deleted) {
+ markLocalMutation(ident);
+ }
catalogCache.invalidate(ident);
return deleted;
@@ -1331,6 +1412,7 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
return newCatalogBuilder.build();
});
+ markLocalMutation(nameIdentifier);
catalogCache.invalidate(nameIdentifier);
} catch (NoSuchCatalogException e) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogListener.java
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogListener.java
new file mode 100644
index 0000000000..bd9d54601b
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogListener.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational;
+
+import java.util.List;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+
+/** Listener for batches consumed from {@code entity_change_log}. */
+@FunctionalInterface
+public interface EntityChangeLogListener {
+
+ /**
+ * Handles a batch of entity changes.
+ *
+ * @param changes the entity changes fetched in one poller cycle
+ */
+ void onEntityChange(List<EntityChangeRecord> changes);
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
new file mode 100644
index 0000000000..5719cea9f6
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Global poller for {@code entity_change_log}.
+ *
+ * <p>The poller owns the single high-water mark for a Gravitino server
process and dispatches each
+ * consumed batch to registered listeners. Listeners should only perform
idempotent local cache
+ * invalidation. The cursor always advances after dispatch regardless of
individual listener
+ * failures, so a faulty listener cannot block other listeners or prevent
pruning.
+ */
+public class EntityChangeLogPoller implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(EntityChangeLogPoller.class);
+
+ /** Max entity-change rows to fetch per poller cycle. */
+ private static final int ENTITY_CHANGE_POLLER_MAX_ROWS = 500;
+
+ private final List<EntityChangeLogListener> listeners = new
CopyOnWriteArrayList<>();
+ private final long pollIntervalSecs;
+ private final long retentionMs;
+ private final long cleanupIntervalMs;
+ private final LongSupplier clockMs;
+
+ private ScheduledExecutorService scheduler;
+ private volatile long entityPollHighWaterId = 0;
+ private volatile long lastCleanupMs = Long.MIN_VALUE;
+
+ /**
+ * Creates an {@link EntityChangeLogPoller}.
+ *
+ * @param pollIntervalSecs interval between successive polling cycles
+ */
+ public EntityChangeLogPoller(long pollIntervalSecs) {
+ this(
+ pollIntervalSecs,
+ TimeUnit.DAYS.toMillis(1),
+ TimeUnit.HOURS.toMillis(1),
+ System::currentTimeMillis);
+ }
+
+ /**
+ * Creates an {@link EntityChangeLogPoller}.
+ *
+ * @param pollIntervalSecs interval between successive polling cycles
+ * @param retentionMs entity change retention in milliseconds, or 0 to
disable cleanup
+ * @param cleanupIntervalMs interval between successive cleanup attempts in
milliseconds
+ */
+ public EntityChangeLogPoller(long pollIntervalSecs, long retentionMs, long
cleanupIntervalMs) {
+ this(pollIntervalSecs, retentionMs, cleanupIntervalMs,
System::currentTimeMillis);
+ }
+
+ @VisibleForTesting
+ EntityChangeLogPoller(
+ long pollIntervalSecs, long retentionMs, long cleanupIntervalMs,
LongSupplier clockMs) {
+ Preconditions.checkArgument(pollIntervalSecs > 0, "pollIntervalSecs must
be positive");
+ Preconditions.checkArgument(retentionMs >= 0, "retentionMs must be
non-negative");
+ Preconditions.checkArgument(cleanupIntervalMs > 0, "cleanupIntervalMs must
be positive");
+ this.pollIntervalSecs = pollIntervalSecs;
+ this.retentionMs = retentionMs;
+ this.cleanupIntervalMs = cleanupIntervalMs;
+ this.clockMs = clockMs;
+ }
+
+ /**
+ * Registers a listener to receive future entity change batches.
+ *
+ * @param listener the listener to register
+ */
+ public void registerListener(EntityChangeLogListener listener) {
+ Preconditions.checkArgument(listener != null, "listener cannot be null");
+ listeners.add(listener);
+ }
+
+ /**
+ * Unregisters a previously registered listener.
+ *
+ * @param listener the listener to unregister
+ */
+ public void unregisterListener(EntityChangeLogListener listener) {
+ Preconditions.checkArgument(listener != null, "listener cannot be null");
+ listeners.remove(listener);
+ }
+
+ /**
+ * Initializes the high-water cursor to the current DB tail and schedules
periodic polling.
+ *
+ * <p>On every start (including restarts), the cursor is set to the current
maximum change ID in
+ * the DB, so historical change records written before this server process
started are NOT
+ * replayed. This is intentional: on startup the local cache is cold, so
there is no stale state
+ * to invalidate. Only changes written after this server started need to be
applied to the warming
+ * cache.
+ */
+ public void start() {
+ entityPollHighWaterId =
+ getOrDefault(
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
+
+ scheduler =
+ Executors.newSingleThreadScheduledExecutor(
+ r -> {
+ Thread t = new Thread(r);
+ t.setName("Gravitino-EntityChangeLogPoller");
+ t.setDaemon(true);
+ return t;
+ });
+ scheduler.scheduleWithFixedDelay(
+ this::pollChanges, pollIntervalSecs, pollIntervalSecs,
TimeUnit.SECONDS);
+ }
+
+ @Override
+ public void close() {
+ if (scheduler != null) {
+ scheduler.shutdown();
+ try {
+ if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+ scheduler.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ scheduler.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ @VisibleForTesting
+ void pollChanges() {
+ try {
+ doPollChanges();
+ } catch (Exception e) {
+ if (handleInterruptIfAny(e, "Entity change poll")) {
+ return;
+ }
+ LOG.warn("Entity change poll failed", e);
+ }
+ }
+
+ private synchronized void doPollChanges() {
+ List<EntityChangeRecord> changes = fetchEntityChanges();
+ if (changes.isEmpty()) {
+ pruneExpiredChangesIfNeeded();
+ return;
+ }
+
+ long maxSeenId = entityPollHighWaterId;
+ for (EntityChangeRecord change : changes) {
+ if (change.getId() > maxSeenId) {
+ maxSeenId = change.getId();
+ }
+ }
+
+ List<EntityChangeRecord> dispatchedChanges =
Collections.unmodifiableList(changes);
+ for (EntityChangeLogListener listener : listeners) {
+ try {
+ listener.onEntityChange(dispatchedChanges);
+ } catch (Exception e) {
+ LOG.warn("Entity change listener {} failed",
listener.getClass().getName(), e);
+ }
+ }
+
+ entityPollHighWaterId = maxSeenId;
+ pruneExpiredChangesIfNeeded();
+ }
+
+ private List<EntityChangeRecord> fetchEntityChanges() {
+ return SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
+ m -> m.selectEntityChanges(entityPollHighWaterId,
ENTITY_CHANGE_POLLER_MAX_ROWS));
+ }
+
+ private static boolean handleInterruptIfAny(Throwable e, String context) {
+ Throwable t = e;
+ while (t != null) {
+ if (t instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ LOG.debug("{} interrupted, stopping poll cycle", context);
+ return true;
+ }
+ t = t.getCause();
+ }
+ if (Thread.currentThread().isInterrupted()) {
+ LOG.debug("{} ran while thread was interrupted, stopping poll cycle",
context);
+ return true;
+ }
+ return false;
+ }
+
+ private void pruneExpiredChangesIfNeeded() {
+ if (retentionMs <= 0) {
+ return;
+ }
+
+ long now = clockMs.getAsLong();
+ if (lastCleanupMs != Long.MIN_VALUE && now - lastCleanupMs <
cleanupIntervalMs) {
+ return;
+ }
+
+ long before = now - retentionMs;
+ try {
+ SessionUtils.doWithoutCommit(
+ EntityChangeLogMapper.class, mapper ->
mapper.pruneOldEntityChanges(before));
+ } catch (Exception e) {
+ LOG.warn("Failed to prune expired entity change logs before {}", before,
e);
+ } finally {
+ // Always advance the cursor regardless of success or failure. A
transient DB error
+ // should not cause repeated prune attempts on every poll cycle (every
few seconds)
+ // until one eventually succeeds — the next cleanup will happen after
cleanupIntervalMs.
+ lastCleanupMs = now;
+ }
+ }
+
+ private static long getOrDefault(Long value) {
+ return value == null ? 0L : value;
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
index 27381659d6..030700eb7e 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -28,6 +28,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.Config;
@@ -57,13 +58,15 @@ import org.slf4j.LoggerFactory;
* MySQL, PostgreSQL, etc. If you want to use a different backend, you can
implement the {@link
* RelationalBackend} interface. The default JDBC backend is {@link
JDBCBackend}.
*/
-public class RelationalEntityStore implements EntityStore,
SupportsRelationOperations {
+public class RelationalEntityStore
+ implements EntityStore, SupportsRelationOperations,
SupportsEntityChangeLog {
private static final Logger LOGGER =
LoggerFactory.getLogger(RelationalEntityStore.class);
public static final ImmutableMap<String, String> RELATIONAL_BACKENDS =
ImmutableMap.of(
Configs.DEFAULT_ENTITY_RELATIONAL_STORE,
JDBCBackend.class.getCanonicalName());
private RelationalBackend backend;
private RelationalGarbageCollector garbageCollector;
+ private EntityChangeLogPoller entityChangeLogPoller;
private EntityCache cache;
@VisibleForTesting
@@ -85,6 +88,17 @@ public class RelationalEntityStore implements EntityStore,
SupportsRelationOpera
this.backend = createRelationalEntityBackend(config);
this.garbageCollector = new RelationalGarbageCollector(backend, config);
this.garbageCollector.start();
+
+ // The change-log poller is a side module of the entity store: it polls
the entity_change_log
+ // table this store writes to, dispatches batches to registered listeners
(e.g. for cross-node
+ // cache invalidation), and prunes expired rows. Like the garbage
collector, it is owned and
+ // lifecycle-managed by the store itself.
+ this.entityChangeLogPoller =
+ new EntityChangeLogPoller(
+ config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS),
+
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)),
+
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)));
+ this.entityChangeLogPoller.start();
}
private RelationalBackend createRelationalEntityBackend(Config config) {
@@ -198,9 +212,20 @@ public class RelationalEntityStore implements EntityStore,
SupportsRelationOpera
throw new UnsupportedOperationException("Unsupported operation in
relational entity store.");
}
+ @Override
+ public void registerEntityChangeLogListener(EntityChangeLogListener
listener) {
+ entityChangeLogPoller.registerListener(listener);
+ }
+
+ @Override
+ public void unregisterEntityChangeLogListener(EntityChangeLogListener
listener) {
+ entityChangeLogPoller.unregisterListener(listener);
+ }
+
@Override
public void close() throws IOException {
cache.clear();
+ entityChangeLogPoller.close();
garbageCollector.close();
backend.close();
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/SupportsEntityChangeLog.java
b/core/src/main/java/org/apache/gravitino/storage/relational/SupportsEntityChangeLog.java
new file mode 100644
index 0000000000..02bed6d308
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/SupportsEntityChangeLog.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational;
+
+/**
+ * Capability implemented by an {@link org.apache.gravitino.EntityStore} that
maintains an {@code
+ * entity_change_log} and runs a background poller dispatching consumed change
batches to registered
+ * listeners.
+ *
+ * <p>The poller is a side module owned and lifecycle-managed by the entity
store itself (created
+ * and started when the store is initialized, stopped when the store is
closed). Consumers that need
+ * to react to entity changes — e.g. for cross-node cache invalidation —
register a listener through
+ * this capability rather than reaching for a global component.
+ */
+public interface SupportsEntityChangeLog {
+
+ /**
+ * Registers a listener to receive future entity change batches.
+ *
+ * @param listener the listener to register
+ */
+ default void registerEntityChangeLogListener(EntityChangeLogListener
listener) {
+ // default no-op implementation since not all stores will support this
capability
+ }
+
+ /**
+ * Unregisters a previously registered listener.
+ *
+ * @param listener the listener to unregister
+ */
+ default void unregisterEntityChangeLogListener(EntityChangeLogListener
listener) {
+ // default no-op implementation since not all stores will support this
capability
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
index 34e8ca13db..a85e1b85ba 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/CatalogMetaService.java
@@ -225,8 +225,6 @@ public class CatalogMetaService {
String metalakeName = identifier.namespace().level(0);
String oldFullName =
NameIdentifierUtil.ofCatalog(metalakeName,
oldCatalogEntity.name()).toString();
- String newFullName = NameIdentifierUtil.ofCatalog(metalakeName,
newEntity.name()).toString();
- boolean isRenamed = !Objects.equals(oldFullName, newFullName);
AtomicInteger updateResult = new AtomicInteger(0);
try {
@@ -241,7 +239,7 @@ public class CatalogMetaService {
oldCatalogPO, newEntity,
oldCatalogPO.getMetalakeId()),
oldCatalogPO))),
() -> {
- if (isRenamed && updateResult.get() > 0) {
+ if (updateResult.get() > 0) {
SessionUtils.doWithoutCommit(
EntityChangeLogMapper.class,
mapper ->
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
index 0928d5d9f9..e3523d3bd1 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
@@ -20,6 +20,9 @@ package org.apache.gravitino.authorization;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
@@ -128,6 +131,9 @@ public class TestAccessControlManager {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
Mockito.when(config.get(CATALOG_CACHE_EVICTION_INTERVAL_MS)).thenReturn(1000L);
// Fix cache for testing.
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
index 5226e5d13d..936d5336a7 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
@@ -20,6 +20,9 @@ package org.apache.gravitino.authorization;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
@@ -110,6 +113,9 @@ public class TestOwnerManager {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
Mockito.when(config.get(CATALOG_CACHE_EVICTION_INTERVAL_MS)).thenReturn(1000L);
// Fix the cache config for testing
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 5582bcb027..5d978dd3b0 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -35,9 +35,11 @@ import com.google.common.collect.Sets;
import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.CatalogChange;
@@ -65,6 +67,10 @@ import org.apache.gravitino.meta.SchemaVersion;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.memory.TestMemoryEntityStore;
import
org.apache.gravitino.storage.memory.TestMemoryEntityStore.InMemoryEntityStore;
+import org.apache.gravitino.storage.relational.EntityChangeLogListener;
+import org.apache.gravitino.storage.relational.SupportsEntityChangeLog;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.utils.PrincipalUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
@@ -604,6 +610,156 @@ public class TestCatalogManager {
.createCatalogWrapper(any(CatalogEntity.class), eq(null));
}
+ @Test
+ void testCatalogChangeLogListenerInvalidatesCatalogCacheForRemoteChange()
throws Exception {
+ NameIdentifier ident = NameIdentifier.of("metalake", "change_log_catalog");
+ Map<String, String> props =
+ ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+
+ catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider,
"comment", props);
+ Assertions.assertNotNull(catalogManager.loadCatalogAndWrap(ident));
+
Assertions.assertNotNull(catalogManager.getCatalogCache().getIfPresent(ident));
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+ listener.onEntityChange(
+ List.of(
+ new EntityChangeRecord(
+ 1L, "metalake", "CATALOG", "metalake.change_log_catalog",
OperateType.ALTER, 0L)));
+
+
Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(ident));
+ }
+
+ @Test
+ void testCatalogChangeLogListenerSkipsInvalidationForLocalMutation() throws
Exception {
+ NameIdentifier ident = NameIdentifier.of("metalake", "change_log_local");
+ Map<String, String> props =
+ ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+
+ catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider,
"comment", props);
+ Assertions.assertNotNull(catalogManager.loadCatalogAndWrap(ident));
+
Assertions.assertNotNull(catalogManager.getCatalogCache().getIfPresent(ident));
+
+ // Enable local-mutation tracking, which a CatalogManager backed by a
change-log-aware store
+ // would have turned on in its constructor. The in-memory store used here
does not support a
+ // change log, so set it explicitly.
+ catalogManager.setTrackLocalMutations(true);
+ catalogManager.markLocalMutation(ident);
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+ listener.onEntityChange(
+ List.of(
+ new EntityChangeRecord(
+ 1L, "metalake", "CATALOG", "metalake.change_log_local",
OperateType.ALTER, 0L)));
+
+ Assertions.assertNotNull(
+ catalogManager.getCatalogCache().getIfPresent(ident),
+ "Cache should NOT be invalidated for local mutations");
+ }
+
+ @Test
+ void
testCatalogChangeLogListenerSkipsBadRecordAndStillProcessesLaterValidChange()
+ throws Exception {
+ NameIdentifier ident = NameIdentifier.of("metalake", "change_log_batch");
+ Map<String, String> props =
+ ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+
+ catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider,
"comment", props);
+ Assertions.assertNotNull(catalogManager.loadCatalogAndWrap(ident));
+
Assertions.assertNotNull(catalogManager.getCatalogCache().getIfPresent(ident));
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+ listener.onEntityChange(
+ List.of(
+ new EntityChangeRecord(
+ 1L, "metalake", null, "metalake.change_log_batch",
OperateType.ALTER, 0L),
+ new EntityChangeRecord(
+ 2L, "metalake", "CATALOG", "metalake.change_log_batch",
OperateType.ALTER, 0L)));
+
+ Assertions.assertNull(
+ catalogManager.getCatalogCache().getIfPresent(ident),
+ "Cache should still be invalidated by the later valid record");
+ }
+
+ @Test
+ void testCloseUnregistersCatalogChangeLogListener() {
+ ChangeLogAwareEntityStore store = new ChangeLogAwareEntityStore();
+ CatalogManager manager = new CatalogManager(config, store, new
RandomIdGenerator());
+
+ EntityChangeLogListener registeredListener = store.listener.get();
+ Assertions.assertNotNull(registeredListener);
+
+ manager.close();
+
+ Assertions.assertSame(registeredListener,
store.unregisteredListener.get());
+ }
+
+ @Test
+ void testDropCatalogDoesNotMarkLocalMutationWhenStoreReturnsFalse() throws
Exception {
+ ChangeLogAwareEntityStore store = new ChangeLogAwareEntityStore();
+ store.initialize(config);
+ store.put(metalakeEntity, true);
+
+ CatalogManager manager = new CatalogManager(config, store, new
RandomIdGenerator());
+ NameIdentifier ident = NameIdentifier.of("metalake",
"delete_returns_false");
+ Map<String, String> props =
+ ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+
+ manager.createCatalog(ident, Catalog.Type.RELATIONAL, provider, "comment",
props);
+ store.returnFalseForCatalogDelete = true;
+
+ Assertions.assertFalse(manager.dropCatalog(ident, true));
+ Assertions.assertNotNull(manager.loadCatalogAndWrap(ident));
+ Assertions.assertNotNull(manager.getCatalogCache().getIfPresent(ident));
+
+ store
+ .listener
+ .get()
+ .onEntityChange(
+ List.of(
+ new EntityChangeRecord(
+ 1L,
+ "metalake",
+ "CATALOG",
+ "metalake.delete_returns_false",
+ OperateType.ALTER,
+ 0L)));
+
+ Assertions.assertNull(manager.getCatalogCache().getIfPresent(ident));
+ manager.close();
+ }
+
@Test
public void testDropCatalogSkipsImportedSchemas() throws Exception {
NameIdentifier ident = NameIdentifier.of("metalake", "test41");
@@ -661,6 +817,33 @@ public class TestCatalogManager {
Assertions.assertTrue(catalogManager.dropCatalog(ident));
}
+ private static class ChangeLogAwareEntityStore extends InMemoryEntityStore
+ implements SupportsEntityChangeLog {
+ private final AtomicReference<EntityChangeLogListener> listener = new
AtomicReference<>();
+ private final AtomicReference<EntityChangeLogListener>
unregisteredListener =
+ new AtomicReference<>();
+ private boolean returnFalseForCatalogDelete;
+
+ @Override
+ public boolean delete(NameIdentifier ident, EntityType entityType, boolean
cascade)
+ throws IOException {
+ if (returnFalseForCatalogDelete && entityType == EntityType.CATALOG) {
+ return false;
+ }
+ return super.delete(ident, entityType, cascade);
+ }
+
+ @Override
+ public void registerEntityChangeLogListener(EntityChangeLogListener
listener) {
+ this.listener.set(listener);
+ }
+
+ @Override
+ public void unregisterEntityChangeLogListener(EntityChangeLogListener
listener) {
+ this.unregisteredListener.set(listener);
+ }
+ }
+
@Test
public void testDropCatalogIgnoresMissingSchema() throws Exception {
NameIdentifier ident = NameIdentifier.of("metalake", "test41");
diff --git
a/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
b/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
index 9ea21d23e7..0a97ec1efb 100644
---
a/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
@@ -20,6 +20,9 @@ package org.apache.gravitino.hook;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
@@ -214,6 +217,9 @@ public class TestFilesetHookDispatcher extends
TestOperationDispatcher {
.thenReturn(1000L);
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60
* 1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
Mockito.when(config.get(CATALOG_CACHE_EVICTION_INTERVAL_MS)).thenReturn(1000L);
Mockito.doReturn(100000L).when(config).get(TREE_LOCK_MAX_NODE_IN_MEMORY);
diff --git
a/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
b/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
index fe5d5d05b7..45f2cc4ca4 100644
--- a/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
+++ b/core/src/test/java/org/apache/gravitino/hook/TestTableHookDispatcher.java
@@ -20,6 +20,9 @@ package org.apache.gravitino.hook;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
@@ -174,6 +177,9 @@ public class TestTableHookDispatcher extends
TestOperationDispatcher {
.thenReturn(1000L);
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60
* 1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
Mockito.when(config.get(CATALOG_CACHE_EVICTION_INTERVAL_MS)).thenReturn(1000L);
Mockito.doReturn(100000L).when(config).get(TREE_LOCK_MAX_NODE_IN_MEMORY);
diff --git
a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
index 91441d2a3b..5b80249042 100644
--- a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
+++ b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
@@ -20,6 +20,9 @@
package org.apache.gravitino.policy;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
@@ -195,6 +198,9 @@ public class TestPolicyManager {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
diff --git
a/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
b/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
index 40d9ad90bd..aa0cb0fea5 100644
--- a/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
+++ b/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
@@ -20,6 +20,9 @@
package org.apache.gravitino.stats;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
@@ -109,6 +112,9 @@ public class TestStatisticManager {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
b/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
index ece352f8ef..cac53eed61 100644
---
a/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
+++
b/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
@@ -20,6 +20,9 @@
package org.apache.gravitino.storage;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD;
@@ -143,6 +146,9 @@ abstract class AbstractEntityStorageTest {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS)).thenReturn(100);
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java
new file mode 100644
index 0000000000..4eea24e24f
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.apache.gravitino.storage.relational.po.cache.OperateType;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+public class TestEntityChangeLogPoller {
+
+ @Test
+ void testRejectsNonPositivePollInterval() {
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(0));
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(-1));
+ }
+
+ @Test
+ void testPollChangesDispatchesSameBatchToAllListenersAndAdvancesCursor() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
+ EntityChangeRecord second = change(2L, "SCHEMA", "ml1.cat1.sch1");
+ when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(first,
second));
+ when(mapper.selectEntityChanges(2L, 500)).thenReturn(List.of());
+
+ List<EntityChangeRecord> firstListenerRecords = new ArrayList<>();
+ List<EntityChangeRecord> secondListenerRecords = new ArrayList<>();
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ sessionUtils
+ .when(() -> SessionUtils.getWithoutCommit(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<Object, Object> func = invocation.getArgument(1);
+ return func.apply(mapper);
+ });
+
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ poller.registerListener(firstListenerRecords::addAll);
+ poller.registerListener(secondListenerRecords::addAll);
+
+ poller.pollChanges();
+ poller.pollChanges();
+ }
+
+ Assertions.assertEquals(List.of(first, second), firstListenerRecords);
+ Assertions.assertEquals(List.of(first, second), secondListenerRecords);
+ }
+
+ @Test
+ void testListenerFailureDoesNotBlockOtherListenersAndCursorStillAdvances() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
+ when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(change));
+ when(mapper.selectEntityChanges(1L, 500)).thenReturn(List.of());
+
+ List<EntityChangeRecord> received = new ArrayList<>();
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ sessionUtils
+ .when(() -> SessionUtils.getWithoutCommit(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<Object, Object> func = invocation.getArgument(1);
+ return func.apply(mapper);
+ });
+
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ poller.registerListener(
+ changes -> {
+ throw new RuntimeException("listener failed");
+ });
+ poller.registerListener(received::addAll);
+
+ poller.pollChanges();
+ poller.pollChanges();
+ }
+
+ Assertions.assertEquals(List.of(change), received);
+ }
+
+ @Test
+ void testPollChangesCatchesFetchFailures() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 500)).thenThrow(new
RuntimeException("db failed"));
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ sessionUtils
+ .when(() -> SessionUtils.getWithoutCommit(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<Object, Object> func = invocation.getArgument(1);
+ return func.apply(mapper);
+ });
+
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+
+ Assertions.assertDoesNotThrow(poller::pollChanges);
+ }
+ }
+
+ @Test
+ void testDispatchesImmutableBatchToListeners() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
+ EntityChangeRecord second = change(2L, "SCHEMA", "ml1.cat1.sch1");
+ when(mapper.selectEntityChanges(0L, 500)).thenReturn(new
ArrayList<>(List.of(first, second)));
+
+ List<EntityChangeRecord> received = new ArrayList<>();
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ sessionUtils
+ .when(() -> SessionUtils.getWithoutCommit(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<Object, Object> func = invocation.getArgument(1);
+ return func.apply(mapper);
+ });
+ sessionUtils
+ .when(() -> SessionUtils.doWithoutCommit(any(), any()))
+ .thenAnswer(invocation -> null);
+
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ poller.registerListener(
+ changes ->
Assertions.assertThrows(UnsupportedOperationException.class, changes::clear));
+ poller.registerListener(received::addAll);
+
+ poller.pollChanges();
+ }
+
+ Assertions.assertEquals(List.of(first, second), received);
+ }
+
+ @Test
+ void testPrunesExpiredChangesAfterCleanupInterval() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of());
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller =
+ new EntityChangeLogPoller(
+ 1, TimeUnit.DAYS.toMillis(1), TimeUnit.HOURS.toMillis(1), () ->
100_000_000L);
+
+ poller.pollChanges();
+ }
+
+ verify(mapper).pruneOldEntityChanges(100_000_000L -
TimeUnit.DAYS.toMillis(1));
+ }
+
+ @Test
+ void testSkipsPruneBeforeCleanupInterval() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of());
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller =
+ new EntityChangeLogPoller(
+ 1, TimeUnit.DAYS.toMillis(1), TimeUnit.HOURS.toMillis(1), () ->
100_000_000L);
+
+ poller.pollChanges();
+ poller.pollChanges();
+ }
+
+ verify(mapper).pruneOldEntityChanges(100_000_000L -
TimeUnit.DAYS.toMillis(1));
+ }
+
+ @Test
+ void testDisablesPruneWhenRetentionIsZero() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of());
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller =
+ new EntityChangeLogPoller(1, 0L, TimeUnit.HOURS.toMillis(1), () ->
100_000_000L);
+
+ poller.pollChanges();
+ }
+
+ verify(mapper, never()).pruneOldEntityChanges(anyLong());
+ }
+
+ private static EntityChangeRecord change(long id, String type, String
fullName) {
+ return new EntityChangeRecord(id, "ml1", type, fullName,
OperateType.ALTER, 0L);
+ }
+
+ private static void mockSessionUtils(
+ MockedStatic<SessionUtils> sessionUtils, EntityChangeLogMapper mapper) {
+ sessionUtils
+ .when(() -> SessionUtils.getWithoutCommit(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<Object, Object> func = invocation.getArgument(1);
+ return func.apply(mapper);
+ });
+ sessionUtils
+ .when(() -> SessionUtils.doWithoutCommit(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Consumer<Object> consumer = invocation.getArgument(1);
+ consumer.accept(mapper);
+ return null;
+ });
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
index 7d85ea7f99..f58c6b9da3 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.storage.relational.service;
import java.io.IOException;
import java.util.List;
import java.util.Map;
+import org.apache.gravitino.Catalog;
import org.apache.gravitino.Entity;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.meta.BaseMetalake;
@@ -109,14 +110,40 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
createAndInsertMakeLake(METALAKE_NAME);
CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
CATALOG_NAME);
+ long maxIdBeforeCatalogAlter = maxEntityChangeId();
+ CatalogEntity alteredCatalog =
+ backend.update(
+ catalog.nameIdentifier(),
+ Entity.EntityType.CATALOG,
+ entity ->
+ CatalogEntity.builder()
+ .withId(catalog.id())
+ .withNamespace(catalog.namespace())
+ .withName(CATALOG_NAME)
+ .withType(Catalog.Type.RELATIONAL)
+ .withProvider("test")
+ .withComment("updated comment")
+ .withProperties(null)
+ .withAuditInfo(AUDIT_INFO)
+ .build());
+ assertEntityChange(
+ maxIdBeforeCatalogAlter,
+ METALAKE_NAME,
+ Entity.EntityType.CATALOG,
+ NameIdentifierUtil.ofCatalog(METALAKE_NAME, CATALOG_NAME).toString(),
+ OperateType.ALTER);
+
long maxIdBeforeCatalogRename = maxEntityChangeId();
CatalogEntity renamedCatalog =
backend.update(
- catalog.nameIdentifier(),
+ alteredCatalog.nameIdentifier(),
Entity.EntityType.CATALOG,
entity ->
createCatalog(
- catalog.id(), catalog.namespace(), CATALOG_NAME +
"_renamed", AUDIT_INFO));
+ alteredCatalog.id(),
+ alteredCatalog.namespace(),
+ CATALOG_NAME + "_renamed",
+ AUDIT_INFO));
assertEntityChange(
maxIdBeforeCatalogRename,
METALAKE_NAME,
diff --git a/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
b/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
index d7e86abba3..ae377b8ad0 100644
--- a/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
+++ b/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
@@ -19,6 +19,9 @@
package org.apache.gravitino.tag;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
@@ -131,6 +134,9 @@ public class TestTagManager {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
// Fix cache config for test
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index 2aec2d9183..035662839b 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -57,6 +57,9 @@ The following table lists the storage configuration items:
| `gravitino.entity.store.maxTransactionSkewTimeMs` | The maximum skew time of
transactions in milliseconds.
| `2000`
| No | 0.3.0
|
| `gravitino.entity.store.deleteAfterTimeMs` | The maximum time in
milliseconds that deleted and old-version data is kept. Set to at least 10
minutes and no longer than 30 days.
|
`604800000`(7 days) | No
| 0.5.0 |
| `gravitino.entity.store.versionRetentionCount` | The Count of versions
allowed to be retained, including the current version, used to delete old
versions data. Set to at least 1 and no greater than 10.
| `1`
| No |
0.5.0 |
+| `gravitino.entityChangeLog.pollIntervalSecs` | The interval in seconds
for polling the entity change log. The poller invalidates stale local caches
(e.g. the catalog cache) across HA nodes by consuming change log records. Must
be positive. | `3`
| No |
1.3.0 |
+| `gravitino.entityChangeLog.retentionSecs` | The retention time in
seconds for entity change log rows. Expired rows are pruned periodically. Set
to `0` to disable automatic cleanup. Must be non-negative.
| `86400`(1
day) | No |
1.3.0 |
+| `gravitino.entityChangeLog.cleanupIntervalSecs` | The interval in seconds
for pruning expired entity change log rows. Must be positive.
| `3600`(1
hour) | No |
1.3.0 |
| `gravitino.entity.store.relational` | Detailed implementation
of Relational storage. `H2`, `MySQL` and `PostgreSQL` is currently supported,
and the implementation is `JDBCBackend`.
|
`JDBCBackend` | No
| 0.5.0 |
| `gravitino.entity.store.relational.jdbcUrl` | The database url that
the `JDBCBackend` needs to connect to. If you use `MySQL` or `PostgreSQL`, you
should firstly initialize the database tables yourself by executing the ddl
scripts in the `${GRAVITINO_HOME}/scripts/{DATABASE_TYPE}/` directory. |
`jdbc:h2` | No
| 0.5.0 |
| `gravitino.entity.store.relational.jdbcDriver` | The jdbc driver name
that the `JDBCBackend` needs to use. You should place the driver Jar package in
the `${GRAVITINO_HOME}/libs/` directory.
|
`org.h2.Driver` | Yes if the jdbc connection url is not
`jdbc:h2` | 0.5.0 |
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizationLookups.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizationLookups.java
index 5ff033f5e1..1e7c5949c0 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizationLookups.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizationLookups.java
@@ -35,8 +35,8 @@ import
org.apache.gravitino.storage.relational.utils.SessionUtils;
* falls back to a shared {@link GravitinoCache} on a request miss, and
finally issues a single DB
* query on a cache miss. A successful DB fetch populates both tiers so
subsequent calls — in this
* request and later ones — hit the cache. The two underlying caches are
invalidated externally by
- * {@link JcasbinChangePoller} (HA peers) and by the {@link
- *
org.apache.gravitino.authorization.GravitinoAuthorizer#handleMetadataOwnerChange}
/ {@link
+ * the global entity change log poller, {@link JcasbinChangeListener} (owner
changes), and by the
+ * {@link
org.apache.gravitino.authorization.GravitinoAuthorizer#handleMetadataOwnerChange}
/ {@link
*
org.apache.gravitino.authorization.GravitinoAuthorizer#handleEntityNameIdMappingChange}
hooks
* (local mutations).
*/
@@ -94,9 +94,8 @@ public class JcasbinAuthorizationLookups {
/**
* Two-tier owner lookup: request-level dedup first, then the shared {@code
ownerRelCache}, and
- * finally a single {@code owner_meta} query. Positive DB fetches populate
both tiers; missing
- * owners are cached only for the current request to avoid pinning a
cross-request negative result
- * through a missed invalidation.
+ * finally a single {@code owner_meta} query. Both positive and negative DB
results populate both
+ * tiers so subsequent calls — within this request and from later requests —
avoid a repeat query.
*/
public Optional<OwnerInfo> resolveOwnerId(
Long metadataId,
@@ -104,16 +103,9 @@ public class JcasbinAuthorizationLookups {
AuthorizationRequestContext requestContext) {
return requestContext.computeOwnerIfAbsent(
metadataId,
- id -> {
- try {
- // Use the cache's atomic loader so concurrent misses on the same
id collapse to one DB
- // query. The loader throws for missing owners so only positive
results land in the
- // long-lived cache; negatives are confined to the per-request map
above.
- return ownerRelCache.get(id, k -> loadOwner(k, metadataType));
- } catch (NoSuchOwnerException e) {
- return Optional.empty();
- }
- });
+ // Use the cache's atomic loader so concurrent misses on the same id
collapse to one DB
+ // query. Both present and absent results are cached so later requests
skip the DB entirely.
+ id -> ownerRelCache.get(id, k -> loadOwner(k, metadataType)));
}
private static Optional<OwnerInfo> loadOwner(Long id, MetadataObject.Type
metadataType) {
@@ -121,15 +113,6 @@ public class JcasbinAuthorizationLookups {
SessionUtils.getWithoutCommit(
OwnerMetaMapper.class,
m -> m.selectOwnerByMetadataObjectIdAndType(id,
metadataType.name()));
- if (ownerInfo == null) {
- throw new NoSuchOwnerException();
- }
- return Optional.of(ownerInfo);
- }
-
- private static final class NoSuchOwnerException extends RuntimeException {
- private NoSuchOwnerException() {
- super(null, null, false, false);
- }
+ return Optional.ofNullable(ownerInfo);
}
}
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
index 0af8e33355..f22e6a39f6 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
@@ -58,6 +58,7 @@ import org.apache.gravitino.cache.GravitinoCache;
import org.apache.gravitino.meta.GroupEntity;
import org.apache.gravitino.meta.RoleEntity;
import org.apache.gravitino.server.authorization.MetadataIdConverter;
+import org.apache.gravitino.storage.relational.SupportsEntityChangeLog;
import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserMetaMapper;
@@ -97,14 +98,13 @@ import org.slf4j.LoggerFactory;
* correctness — TTL eviction only bounds memory. User/group role
snapshots use write-based
* TTLs through {@link CaffeineGravitinoCache}; loaded role policies use
access-based TTLs
* through {@link JcasbinLoadedRolesCache}.
- * <li><b>Eventual-consistency caches</b> — {@link #metadataIdCache} and
{@link #ownerRelCache}. A
- * single background poller ({@link #changePoller}) drains {@code
entity_change_log} and
- * {@code owner_meta} change rows since a high-water-mark cursor and
invalidates the affected
- * keys. Other Gravitino nodes therefore observe ALTER/DROP and owner
changes within one poll
- * interval.
+ * <li><b>Eventual-consistency caches</b> — {@link #metadataIdCache} and
{@link #ownerRelCache}.
+ * The global entity change log poller dispatches {@code
entity_change_log} batches to {@link
+ * #changePoller}, while {@link #changePoller} polls {@code owner_meta}.
Other Gravitino nodes
+ * therefore observe ALTER/DROP and owner changes within one poll
interval.
* </ol>
*
- * <p>The pollers are best-effort and intentionally cheap; see {@link
JcasbinChangePoller} for the
+ * <p>The pollers are best-effort and intentionally cheap; see {@link
JcasbinChangeListener} for the
* contracts they rely on (most notably that {@code
entity_change_log.full_name} is the pre-mutation
* name).
*
@@ -159,7 +159,7 @@ public class JcasbinAuthorizer implements
GravitinoAuthorizer {
private JcasbinAuthorizationLookups lookups;
/** Background HA invalidator for {@link #metadataIdCache} and {@link
#ownerRelCache}. */
- private JcasbinChangePoller changePoller;
+ private JcasbinChangeListener changePoller;
@Override
public void initialize() {
@@ -200,7 +200,11 @@ public class JcasbinAuthorizer implements
GravitinoAuthorizer {
metadataIdCache = new CaffeineGravitinoCache<>(ttlMs, metadataIdCacheSize);
ownerRelCache = new CaffeineGravitinoCache<>(ttlMs, ownerCacheSize);
lookups = new JcasbinAuthorizationLookups(metadataIdCache, ownerRelCache);
- changePoller = new JcasbinChangePoller(metadataIdCache, ownerRelCache,
pollIntervalSecs);
+ changePoller = new JcasbinChangeListener(metadataIdCache, ownerRelCache,
pollIntervalSecs);
+ EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
+ if (entityStore instanceof SupportsEntityChangeLog) {
+ ((SupportsEntityChangeLog)
entityStore).registerEntityChangeLogListener(changePoller);
+ }
changePoller.start();
}
@@ -553,6 +557,10 @@ public class JcasbinAuthorizer implements
GravitinoAuthorizer {
@Override
public void close() throws IOException {
if (changePoller != null) {
+ EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
+ if (entityStore instanceof SupportsEntityChangeLog) {
+ ((SupportsEntityChangeLog)
entityStore).unregisterEntityChangeLogListener(changePoller);
+ }
changePoller.close();
}
if (userRoleCache != null) {
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangePoller.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java
similarity index 86%
rename from
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangePoller.java
rename to
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java
index 2786cd52ac..352dfcb8fd 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangePoller.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java
@@ -34,7 +34,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.MetadataObjects;
import org.apache.gravitino.cache.GravitinoCache;
-import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+import org.apache.gravitino.storage.relational.EntityChangeLogListener;
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.po.auth.ChangedOwnerInfo;
import org.apache.gravitino.storage.relational.po.auth.OwnerInfo;
@@ -47,18 +47,14 @@ import org.slf4j.LoggerFactory;
* Eventual-consistency invalidator for {@link JcasbinAuthorizer}'s {@code
metadataIdCache} and
* {@code ownerRelCache}.
*
- * <p>One scheduled thread drains {@code entity_change_log} and {@code
owner_meta} change rows since
- * a high-water-mark cursor and invalidates the affected keys. Other Gravitino
nodes therefore
- * observe ALTER/DROP and owner changes within one poll interval.
+ * <p>This class polls {@code owner_meta} itself and receives {@code
entity_change_log} batches from
+ * the global entity change log poller.
*
- * <p>Both polls run on every tick — a failure in one does not stop the other.
+ * <p>Other Gravitino nodes therefore observe ALTER/DROP and owner changes
within one poll interval.
*/
-public class JcasbinChangePoller implements AutoCloseable {
+public class JcasbinChangeListener implements EntityChangeLogListener,
AutoCloseable {
- private static final Logger LOG =
LoggerFactory.getLogger(JcasbinChangePoller.class);
-
- /** Max entity-change rows to fetch per poller cycle. */
- private static final int ENTITY_CHANGE_POLLER_MAX_ROWS = 500;
+ private static final Logger LOG =
LoggerFactory.getLogger(JcasbinChangeListener.class);
private final GravitinoCache<String, Long> metadataIdCache;
private final GravitinoCache<Long, Optional<OwnerInfo>> ownerRelCache;
@@ -72,14 +68,13 @@ public class JcasbinChangePoller implements AutoCloseable {
// batch soft-deletes (softDeleteOwnerRelByCatalogId etc.) where many rows
share the same ms.
private volatile long ownerPollHighWaterUpdatedAt = 0;
private volatile long ownerPollHighWaterUpdatedAtId = 0;
- private volatile long entityPollHighWaterId = 0;
/**
* @param metadataIdCache the metadata-id cache to invalidate on entity
changes
* @param ownerRelCache the owner cache to invalidate on owner changes
* @param pollIntervalSecs interval between successive polling cycles
*/
- public JcasbinChangePoller(
+ public JcasbinChangeListener(
GravitinoCache<String, Long> metadataIdCache,
GravitinoCache<Long, Optional<OwnerInfo>> ownerRelCache,
long pollIntervalSecs) {
@@ -105,10 +100,6 @@ public class JcasbinChangePoller implements AutoCloseable {
ownerPollHighWaterUpdatedAt = maxOwnerChange.getUpdatedAt();
ownerPollHighWaterUpdatedAtId = maxOwnerChange.getId();
}
- entityPollHighWaterId =
- getOrDefault(
- SessionUtils.getWithoutCommit(
- EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
scheduler =
Executors.newSingleThreadScheduledExecutor(
@@ -136,16 +127,6 @@ public class JcasbinChangePoller implements AutoCloseable {
}
LOG.warn("Owner change poll failed", e);
}
-
- try {
- LOG.debug("Polling for entity changes after id {}",
entityPollHighWaterId);
- pollEntityChanges();
- } catch (Exception e) {
- if (handleInterruptIfAny(e, "Entity change poll")) {
- return;
- }
- LOG.warn("Entity change poll failed", e);
- }
}
/**
@@ -217,8 +198,7 @@ public class JcasbinChangePoller implements AutoCloseable {
}
/**
- * Drains entity-change rows past {@link #entityPollHighWaterId} and
invalidates the affected
- * {@code metadataIdCache} keys.
+ * Invalidates the affected {@code metadataIdCache} keys from an
entity-change batch.
*
* <p><b>Contract with the writer side:</b> {@code
entity_change_log.full_name} must be the
* <i>pre-mutation</i> name (the name that consumers currently have cached).
The writers in {@code
@@ -231,13 +211,8 @@ public class JcasbinChangePoller implements AutoCloseable {
* for the rationale. The single-threaded scheduler already prevents
overlapping runs in
* production, and the per-batch invalidation atomicity is provided by the
cache itself.
*/
- private synchronized void pollEntityChanges() {
- List<EntityChangeRecord> changes =
- SessionUtils.getWithoutCommit(
- EntityChangeLogMapper.class,
- m -> m.selectEntityChanges(entityPollHighWaterId,
ENTITY_CHANGE_POLLER_MAX_ROWS));
-
- long maxSeenId = entityPollHighWaterId;
+ @Override
+ public synchronized void onEntityChange(List<EntityChangeRecord> changes) {
Set<String> containerPrefixes = new LinkedHashSet<>();
Set<String> leafKeys = new LinkedHashSet<>();
for (EntityChangeRecord change : changes) {
@@ -250,9 +225,6 @@ public class JcasbinChangePoller implements AutoCloseable {
mdType =
MetadataObject.Type.valueOf(entityType.toUpperCase(Locale.ROOT));
} catch (IllegalArgumentException e) {
LOG.warn("Unknown entity type in change log: {}", entityType);
- if (change.getId() > maxSeenId) {
- maxSeenId = change.getId();
- }
continue;
}
@@ -264,13 +236,8 @@ public class JcasbinChangePoller implements AutoCloseable {
} else {
leafKeys.add(cacheKey);
}
-
- if (change.getId() > maxSeenId) {
- maxSeenId = change.getId();
- }
}
invalidateCoalescedKeys(containerPrefixes, leafKeys);
- entityPollHighWaterId = maxSeenId;
}
@Override
diff --git
a/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinAuthorizationLookups.java
b/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinAuthorizationLookups.java
index de0fa232cf..6024c457c1 100644
---
a/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinAuthorizationLookups.java
+++
b/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinAuthorizationLookups.java
@@ -21,6 +21,7 @@ package org.apache.gravitino.server.authorization.jcasbin;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -30,6 +31,7 @@ import java.util.function.Function;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.MetadataObjects;
import org.apache.gravitino.authorization.AuthorizationRequestContext;
+import org.apache.gravitino.cache.CaffeineGravitinoCache;
import org.apache.gravitino.cache.GravitinoCache;
import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.po.auth.OwnerInfo;
@@ -100,7 +102,7 @@ public class TestJcasbinAuthorizationLookups {
}
@Test
- void testResolveOwnerIdDoesNotCacheMissingOwnerInSharedCache() {
+ void testResolveOwnerIdCachesMissingOwnerInSharedCacheWithSameContext() {
CountingCache<String, Long> metadataIdCache = new CountingCache<>(100L);
CountingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
CountingCache<>();
JcasbinAuthorizationLookups lookups =
@@ -116,9 +118,43 @@ public class TestJcasbinAuthorizationLookups {
lookups.resolveOwnerId(100L, MetadataObject.Type.TABLE,
requestContext).isPresent());
}
+ // Shared cache consulted once; second call hits per-request cache.
Assertions.assertEquals(1, ownerRelCache.getCount);
Assertions.assertEquals(0, ownerRelCache.getIfPresentCount);
- Assertions.assertEquals(0, ownerRelCache.putCount);
+ // Absent result is now stored in the shared cache (putCount=1) so later
requests skip the DB.
+ Assertions.assertEquals(1, ownerRelCache.putCount);
+ }
+
+ @Test
+ void testResolveOwnerIdCachesMissingOwnerInSharedCache() {
+ OwnerMetaMapper ownerMetaMapper = mock(OwnerMetaMapper.class);
+ when(ownerMetaMapper.selectOwnerByMetadataObjectIdAndType(100L,
"TABLE")).thenReturn(null);
+
+ CountingCache<String, Long> metadataIdCache = new CountingCache<>(100L);
+ try (CaffeineGravitinoCache<Long, Optional<OwnerInfo>> ownerRelCache =
+ new CaffeineGravitinoCache<>(60_000L, 100L);
+ MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ sessionUtils
+ .when(() -> SessionUtils.getWithoutCommit(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<Object, Object> func = invocation.getArgument(1);
+ return func.apply(ownerMetaMapper);
+ });
+ JcasbinAuthorizationLookups lookups =
+ new JcasbinAuthorizationLookups(metadataIdCache, ownerRelCache);
+
+ Assertions.assertFalse(
+ lookups
+ .resolveOwnerId(100L, MetadataObject.Type.TABLE, new
AuthorizationRequestContext())
+ .isPresent());
+ Assertions.assertFalse(
+ lookups
+ .resolveOwnerId(100L, MetadataObject.Type.TABLE, new
AuthorizationRequestContext())
+ .isPresent());
+ }
+
+ verify(ownerMetaMapper,
times(1)).selectOwnerByMetadataObjectIdAndType(100L, "TABLE");
}
private static class CountingCache<K, V> implements GravitinoCache<K, V> {
diff --git
a/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinChangePoller.java
b/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinChangePoller.java
index 9e9c248167..6bd330fc4c 100644
---
a/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinChangePoller.java
+++
b/server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinChangePoller.java
@@ -18,31 +18,20 @@
*/
package org.apache.gravitino.server.authorization.jcasbin;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockStatic;
-import static org.mockito.Mockito.when;
-
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.Optional;
-import java.util.function.Function;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.cache.GravitinoCache;
-import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
-import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
import org.apache.gravitino.storage.relational.po.auth.OwnerInfo;
import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
-import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import org.mockito.MockedStatic;
-/** Tests for {@link JcasbinChangePoller} static helpers. */
+/** Tests for {@link JcasbinChangeListener} static helpers. */
public class TestJcasbinChangePoller {
@Test
@@ -52,30 +41,30 @@ public class TestJcasbinChangePoller {
Assertions.assertThrows(
IllegalArgumentException.class,
- () -> new JcasbinChangePoller(metadataIdCache, ownerRelCache, 0));
+ () -> new JcasbinChangeListener(metadataIdCache, ownerRelCache, 0));
Assertions.assertThrows(
IllegalArgumentException.class,
- () -> new JcasbinChangePoller(metadataIdCache, ownerRelCache, -1));
+ () -> new JcasbinChangeListener(metadataIdCache, ownerRelCache, -1));
}
@Test
void testChangeLogFullNameStripsLeadingMetalakeForChildTypes() {
MetadataObject catalog =
- JcasbinChangePoller.metadataObjectFromChangeLog(
+ JcasbinChangeListener.metadataObjectFromChangeLog(
"ml1", "ml1.cat1", MetadataObject.Type.CATALOG);
Assertions.assertEquals(
key("ml1", "CATALOG", "cat1", ""),
JcasbinAuthorizationCacheKeys.metadataIdCacheKey("ml1", catalog));
MetadataObject schema =
- JcasbinChangePoller.metadataObjectFromChangeLog(
+ JcasbinChangeListener.metadataObjectFromChangeLog(
"ml1", "ml1.cat1.sch1", MetadataObject.Type.SCHEMA);
Assertions.assertEquals(
key("ml1", "CATALOG", "cat1", "SCHEMA", "sch1", ""),
JcasbinAuthorizationCacheKeys.metadataIdCacheKey("ml1", schema));
MetadataObject table =
- JcasbinChangePoller.metadataObjectFromChangeLog(
+ JcasbinChangeListener.metadataObjectFromChangeLog(
"ml1", "ml1.cat1.sch1.tbl1", MetadataObject.Type.TABLE);
Assertions.assertEquals(
key("ml1", "CATALOG", "cat1", "SCHEMA", "sch1", "TABLE", "tbl1", ""),
@@ -85,7 +74,8 @@ public class TestJcasbinChangePoller {
@Test
void testChangeLogFullNameForMetalakeKeepsItself() {
MetadataObject metalake =
- JcasbinChangePoller.metadataObjectFromChangeLog("ml1", "ml1",
MetadataObject.Type.METALAKE);
+ JcasbinChangeListener.metadataObjectFromChangeLog(
+ "ml1", "ml1", MetadataObject.Type.METALAKE);
Assertions.assertEquals(
key("ml1", "METALAKE", ""),
JcasbinAuthorizationCacheKeys.metadataIdCacheKey("ml1", metalake));
@@ -95,37 +85,14 @@ public class TestJcasbinChangePoller {
void testPollEntityChangesCoalescesContainerPrefixes() {
RecordingCache<String, Long> metadataIdCache = new RecordingCache<>();
RecordingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
RecordingCache<>();
- EntityChangeLogMapper entityChangeLogMapper =
mock(EntityChangeLogMapper.class);
- OwnerMetaMapper ownerMetaMapper = mock(OwnerMetaMapper.class);
-
- when(ownerMetaMapper.selectChangedOwners(0L,
0L)).thenReturn(Collections.emptyList());
- when(entityChangeLogMapper.selectEntityChanges(0L, 500))
- .thenReturn(
- List.of(
- change(1L, MetadataObject.Type.CATALOG, "ml1.cat1"),
- change(2L, MetadataObject.Type.SCHEMA, "ml1.cat1.sch1"),
- change(3L, MetadataObject.Type.TABLE, "ml1.cat1.sch1.tbl1"),
- change(4L, MetadataObject.Type.TABLE, "ml1.cat2.sch1.tbl1")));
-
- try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
- sessionUtils
- .when(() -> SessionUtils.getWithoutCommit(any(), any()))
- .thenAnswer(
- invocation -> {
- Class<?> mapperClass = invocation.getArgument(0);
- Function<Object, Object> func = invocation.getArgument(1);
- if (mapperClass == OwnerMetaMapper.class) {
- return func.apply(ownerMetaMapper);
- }
- if (mapperClass == EntityChangeLogMapper.class) {
- return func.apply(entityChangeLogMapper);
- }
- return null;
- });
-
- JcasbinChangePoller poller = new JcasbinChangePoller(metadataIdCache,
ownerRelCache, 1);
- poller.pollChanges();
- }
+
+ JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
+ poller.onEntityChange(
+ List.of(
+ change(1L, MetadataObject.Type.CATALOG, "ml1.cat1"),
+ change(2L, MetadataObject.Type.SCHEMA, "ml1.cat1.sch1"),
+ change(3L, MetadataObject.Type.TABLE, "ml1.cat1.sch1.tbl1"),
+ change(4L, MetadataObject.Type.TABLE, "ml1.cat2.sch1.tbl1")));
Assertions.assertEquals(
List.of(
@@ -137,11 +104,12 @@ public class TestJcasbinChangePoller {
@Test
void testPollCursorAdvancementIsSynchronized() throws NoSuchMethodException {
- Method pollOwnerChanges =
JcasbinChangePoller.class.getDeclaredMethod("pollOwnerChanges");
- Method pollEntityChanges =
JcasbinChangePoller.class.getDeclaredMethod("pollEntityChanges");
+ Method pollOwnerChanges =
JcasbinChangeListener.class.getDeclaredMethod("pollOwnerChanges");
+ Method onEntityChange =
+ JcasbinChangeListener.class.getDeclaredMethod("onEntityChange",
List.class);
Assertions.assertTrue(Modifier.isSynchronized(pollOwnerChanges.getModifiers()));
-
Assertions.assertTrue(Modifier.isSynchronized(pollEntityChanges.getModifiers()));
+
Assertions.assertTrue(Modifier.isSynchronized(onEntityChange.getModifiers()));
}
private static String key(String... parts) {
