This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 550f701751 [#12440] improvement(core): Replace the entity change log
listener retry/EXIT policy with a cache-clear fallback (#12445)
550f701751 is described below
commit 550f701751dc8c251a9aa0439c39317171c5c1b8
Author: Qi Yu <[email protected]>
AuthorDate: Mon Aug 17 12:06:16 2026 +0800
[#12440] improvement(core): Replace the entity change log listener
retry/EXIT policy with a cache-clear fallback (#12445)
### What changes were proposed in this pull request?
Make "dispatch once, always advance the cursor, listeners are
self-healing" the entity change log contract, and let each listener
recover locally.
Poller and configs:
- `EntityChangeLogPoller`: remove `ListenerFailureAction`,
`exitHandler`, `pendingDelivery`, `BatchDelivery.retryOnly`/`attempts`
and `handleExhaustedRetries`. Each batch is dispatched once and the
cursor always advances; every listener failure is logged at `ERROR`. The
self-healing contract is stated in the poller and
`EntityChangeLogListener` javadoc, so a listener that cannot contain its
own failures is not added silently.
- Remove `gravitino.entityChangeLog.listenerMaxRetries` and
`gravitino.entityChangeLog.listenerFailureAction` from `Configs`, their
wiring in `RelationalEntityStore`, and their entries in
`docs/gravitino-server-config.md`.
- Clean up the removed config stubs in 13 test/benchmark classes.
All three registered listeners now recover by clearing the cache they
maintain, which is a strict superset of the invalidation that failed and
of the rest of the batch:
| Listener | Cache | Recovery |
|-------------------------------|-------------------|--------------------------------------------------|
| `EntityCacheChangeLogListener` | entity cache | clears the whole cache
(already did) |
| `JcasbinChangeListener` | `metadataIdCache` | clears the whole cache
(new) |
| `CatalogChangeLogListener` | catalog cache | clears the whole cache
(new) |
Notes on the two listeners that changed:
- `JcasbinChangeListener` is a third change-log listener that the issue
did not account for. It tolerated poison rows but propagated a failed
invalidation, so under the dispatch-once contract its `metadataIdCache`
would feed a stale name→id mapping to authorization decisions until the
entry's TTL expired.
- `CatalogChangeLogListener` clears the catalog cache on a failed
eviction. This is a deliberate tradeoff, documented in its javadoc:
clearing closes the `CatalogWrapper` of every cached catalog, including
catalogs this process is actively serving, so in-flight requests can hit
`NoClassDefFoundError` from a closed `IsolatedClassLoader` (the failure
mode of #11739). It is accepted so that a changed catalog is never
served stale, and the clear runs only on a failed eviction, off the
normal path. Malformed rows and a failed `consumeLocalMutation` probe
are still skipped rather than escalated, since they name no eviction to
recover.
Caches deliberately left alone: `ownerRelCache` is driven by
`JcasbinChangeListener`'s own `owner_meta` poller, whose cursor only
advances after a successful invalidation batch, so it already retries;
`userRoleCache`, `groupRoleCache` and `loadedRoles` are
version-validated on every read and cannot go stale from a missed batch.
### Why are the changes needed?
After #12374 every registered listener can recover locally, so the
retry/`EXIT` path is effectively unreachable while carrying real cost:
1. `EXIT` trades the whole server for a condition a local cache clear
already resolves. Killing a node to fix a stale cache entry is a heavy,
surprising failure mode for operators.
2. A paused cursor blocks cache invalidation for **every** listener in
the process while one listener retries, so a single misbehaving listener
degrades cluster-wide coherence for up to 10 poll intervals.
3. The retained batch, `pendingDelivery`, `BatchDelivery.retryOnly`,
`attempts` tracking and `handleExhaustedRetries` add machinery and two
public configs for a path no listener reaches.
Fix: #12440
### Does this PR introduce _any_ user-facing change?
Yes:
- Removed config keys `gravitino.entityChangeLog.listenerMaxRetries` and
`gravitino.entityChangeLog.listenerFailureAction`. Both are
`VERSION_2_0_0` and 2.0.0 is unreleased, so no deprecation cycle is
needed.
- A node no longer stops itself (`System.exit(1)`) when a listener keeps
failing to apply a change log batch.
### How was this patch tested?
New and reworked unit tests:
- `TestEntityChangeLogPoller`: the four retry/pause/EXIT/SKIP cases are
replaced by
`testThrowingListenerNeitherPausesCursorNorBlocksOtherListeners` (each
batch dispatched exactly once, the healthy listener sees every batch,
the cursor advances past both) and `testUnregisteredListenerIsSkipped`.
- `TestJcasbinChangePoller` (7 → 14): the `metadataIdCache` clear
fallback on prefix, leaf-key and batch-lock failures; a failed clear
propagating to the poller; the happy path clearing nothing;
`ownerRelCache` not cleared as collateral; plus leaf-vs-prefix keying,
which had no coverage.
- `TestCatalogChangeLogListener` (3 → 7): the clear on a failed
eviction; no clear on the happy path; malformed rows and a failed
`consumeLocalMutation` probe skipped without clearing; a failed clear
propagating.
Suites run locally: `:core:test` (1657 tests) and `:server-common:test`
(272 tests) with `--rerun-tasks`, both green, plus the unit tests of the
four catalog modules whose config stubs changed. `:core:javadoc` reports
no new warnings on the touched files. Docker was not available locally,
so docker-tagged tests and integration tests were not run.
---
.../fileset/TestFilesetCatalogOperations.java | 4 -
.../catalog/kafka/TestKafkaCatalogOperations.java | 4 -
.../generic/TestGenericCatalogOperations.java | 4 -
.../catalog/model/TestModelCatalogOperations.java | 4 -
.../cache/it/AbstractEntityStorageBenchmark.java | 4 -
.../main/java/org/apache/gravitino/Configs.java | 29 ---
.../catalog/CatalogChangeLogListener.java | 137 +++++++++----
.../relational/EntityCacheChangeLogListener.java | 5 +-
.../storage/relational/EntityChangeLogCleaner.java | 2 +-
.../relational/EntityChangeLogListener.java | 8 +-
.../storage/relational/EntityChangeLogPoller.java | 216 ++++++---------------
.../storage/relational/RelationalEntityStore.java | 9 +-
.../authorization/TestAccessControlManager.java | 4 -
.../gravitino/authorization/TestOwnerManager.java | 4 -
.../catalog/TestCatalogChangeLogListener.java | 171 ++++++++++++++--
.../gravitino/hook/TestFilesetHookDispatcher.java | 4 -
.../apache/gravitino/policy/TestPolicyManager.java | 4 -
.../gravitino/stats/TestStatisticManager.java | 4 -
.../storage/AbstractEntityStorageTest.java | 4 -
.../TestEntityCacheCrossNodeInvalidation.java | 7 +-
.../relational/TestEntityChangeLogPoller.java | 173 +++++------------
...TestRelationalEntityStoreHierarchicalCache.java | 2 -
.../org/apache/gravitino/tag/TestTagManager.java | 4 -
docs/gravitino-server-config.md | 12 +-
.../jcasbin/JcasbinChangeListener.java | 64 ++++--
.../jcasbin/TestJcasbinChangePoller.java | 140 ++++++++++++-
26 files changed, 561 insertions(+), 462 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 d8a4f77c13..fe610e7f0f 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
@@ -20,8 +20,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -256,8 +254,6 @@ public class TestFilesetCatalogOperations {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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
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 dd30df96f3..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
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -165,8 +163,6 @@ public class TestKafkaCatalogOperations extends
KafkaClusterEmbedded {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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
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 c1baa184bf..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
@@ -20,8 +20,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -110,8 +108,6 @@ public class TestGenericCatalogOperations {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
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 a95892b088..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
@@ -20,8 +20,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -129,8 +127,6 @@ public class TestModelCatalogOperations {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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
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 3fa058e3b5..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
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -296,8 +294,6 @@ public class AbstractEntityStorageBenchmark<E extends
Entity & HasIdentifier> {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index fb1b0a0c7c..4ed70e1bec 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -20,7 +20,6 @@ package org.apache.gravitino;
import com.google.common.collect.Lists;
import java.io.File;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
@@ -30,7 +29,6 @@ import org.apache.gravitino.config.ConfigBuilder;
import org.apache.gravitino.config.ConfigConstants;
import org.apache.gravitino.config.ConfigEntry;
import org.apache.gravitino.stats.storage.JdbcPartitionStatisticStorageFactory;
-import org.apache.gravitino.storage.relational.EntityChangeLogPoller;
import org.apache.gravitino.utils.FileFetcher;
import org.apache.gravitino.utils.HierarchicalSchemaUtil;
@@ -188,8 +186,6 @@ public class Configs {
.createWithDefault(60 * 60 * 1000L);
public static final long DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS = 3L;
- public static final int DEFAULT_ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES = 10;
- public static final String DEFAULT_ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION
= "EXIT";
public static final long DEFAULT_ENTITY_CHANGE_LOG_RETENTION_SECS = 30 * 24
* 60 * 60L;
public static final long DEFAULT_ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS =
24 * 60 * 60L;
@@ -201,31 +197,6 @@ public class Configs {
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS);
- public static final ConfigEntry<Integer>
ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES =
- new ConfigBuilder("gravitino.entityChangeLog.listenerMaxRetries")
- .doc(
- "The number of times the poller retries a change log batch for a
failing listener"
- + " before applying
gravitino.entityChangeLog.listenerFailureAction")
- .version(ConfigConstants.VERSION_2_0_0)
- .intConf()
- .checkValue(value -> value >= 0,
ConfigConstants.NON_NEGATIVE_NUMBER_ERROR_MSG)
- .createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES);
-
- public static final ConfigEntry<String>
ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION =
- new ConfigBuilder("gravitino.entityChangeLog.listenerFailureAction")
- .doc(
- "What the poller does when a listener exhausted its retries:
EXIT stops this server"
- + " because its local caches are known to be stale, SKIP
drops the batch for that"
- + " listener and keeps serving")
- .version(ConfigConstants.VERSION_2_0_0)
- .stringConf()
- .checkValue(
- value ->
-
Arrays.stream(EntityChangeLogPoller.ListenerFailureAction.values())
- .anyMatch(action ->
action.name().equalsIgnoreCase(value)),
- "The value must be either EXIT or SKIP")
-
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION);
-
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")
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
index 9aa3c843c6..38d9807a0e 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.catalog;
+import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Optional;
@@ -35,12 +36,27 @@ import org.slf4j.LoggerFactory;
* <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.
*
- * <p>This listener never propagates a failure to the poller, so the poller
never retries a batch
- * for it. That is deliberate: local-mutation de-duplication ({@link
- * CatalogManager#consumeLocalMutation}) is single-shot, so re-delivering an
already-applied batch
- * would invalidate a catalog this process mutated itself and close its
still-in-use {@code
- * IsolatedClassLoader}. Dropping an invalidation is the cheaper failure: the
catalog cache expires
- * on access, so staleness is bounded by {@code
gravitino.catalog.cache.evictionIntervalMs}.
+ * <p>The poller hands each batch to a listener only once, so a listener has
to clean up after
+ * itself when it fails. This one does what {@code
EntityCacheChangeLogListener} and {@code
+ * JcasbinChangeListener} do: if removing one catalog from the cache fails, it
clears the whole
+ * catalog cache, which also covers the entry it failed to remove and the rest
of the batch. A row
+ * that cannot be parsed is simply skipped, because it does not point at any
catalog and so cannot
+ * leave anything stale.
+ *
+ * <p>Before removing anything, the listener first goes through the whole
batch and marks off the
+ * changes this node made itself. Doing it in that order means a later failure
cannot leave one of
+ * those marks behind, which would otherwise make a future change from another
node look like a
+ * local one. If the clear itself fails, the exception goes up to the poller,
which logs it at
+ * {@code ERROR} and moves on, and the catalog stays stale until it expires.
+ *
+ * <p><b>What clearing costs:</b> dropping a catalog from the cache closes its
{@code
+ * CatalogWrapper}, which shuts down its connection pool and its {@code
IsolatedClassLoader}.
+ * Clearing the whole cache therefore also closes catalogs this process is
serving right now, and
+ * requests still using classes from a closed classloader can fail with {@code
NoClassDefFoundError}
+ * (that is the bug in #11739). We accept this on purpose so a changed catalog
is never served from
+ * a stale cache; without it, this node would keep serving the old catalog for
up to {@code
+ * gravitino.catalog.cache.evictionIntervalMs}. The clear only happens when a
normal removal failed,
+ * never during normal operation.
*/
public class CatalogChangeLogListener implements EntityChangeLogListener {
@@ -59,51 +75,87 @@ public class CatalogChangeLogListener implements
EntityChangeLogListener {
@Override
public void onEntityChange(List<EntityChangeRecord> changes) {
+ List<CatalogInvalidation> remoteInvalidations = new ArrayList<>();
for (EntityChangeRecord change : changes) {
+ if (!isCatalogChange(change)) {
+ continue;
+ }
+
+ Optional<NameIdentifier> identOpt = catalogIdentifier(change);
+ if (identOpt.isEmpty()) {
+ // Already logged. This row does not point at any catalog, so there is
nothing stale to
+ // clean up. Just skip it instead of clearing the cache.
+ continue;
+ }
+ NameIdentifier ident = identOpt.get();
+
+ boolean localMutation;
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: {},
change log id {}",
- ident,
- change.getId());
- continue;
- }
-
- // Logged at INFO on purpose: this tears down the cached catalog,
including its connection
- // pool and isolated classloader, and it is the main cross-node effect
of the change log.
- // CatalogManager logs the matching "Closing catalog" line when the
eviction runs.
- LOG.info(
- "Invalidating catalog cache for {} due to a remote {} recorded in
change log id {}",
+ localMutation = catalogManager.consumeLocalMutation(ident);
+ } catch (RuntimeException e) {
+ // We could not tell whether this change came from this node or
another one. The name is
+ // valid, so assume it came from another node: the worst case is one
extra cache removal,
+ // while skipping it could leave an old catalog cached forever.
+ LOG.error(
+ "Failed to check local mutation state for catalog {}, treating
change log record id {} "
+ + "as remote to avoid serving stale metadata",
+ ident,
+ change.getId(),
+ e);
+ localMutation = false;
+ }
+
+ if (localMutation) {
+ LOG.debug(
+ "Skipping catalog cache invalidation for local mutation: {},
change log id {}",
ident,
- change.getOperateType(),
change.getId());
+ continue;
+ }
+
+ remoteInvalidations.add(new CatalogInvalidation(change, ident));
+ }
+
+ for (CatalogInvalidation invalidation : remoteInvalidations) {
+ EntityChangeRecord change = invalidation.change;
+ NameIdentifier ident = invalidation.ident;
+ // INFO on purpose: dropping the catalog from the cache also closes its
connection pool and
+ // its isolated classloader, and this is the main thing the change log
does across nodes.
+ // CatalogManager prints the matching "Closing catalog" line when the
removal happens.
+ LOG.info(
+ "Invalidating catalog cache for {} due to a remote {} recorded in
change log id {}",
+ ident,
+ change.getOperateType(),
+ change.getId());
+
+ try {
catalogManager.getCatalogCache().invalidate(ident);
} catch (RuntimeException e) {
- // Deliberately not rethrown: see the class javadoc. A dropped
invalidation only costs
- // bounded staleness here, while a retry of an already-applied batch
can tear down a
- // catalog that is still in use.
- LOG.warn(
- "Failed to process catalog change log record: id={}, fullName={},
entityType={}, "
- + "operateType={}",
+ // This batch will never be sent again, so giving up here would keep
serving the old
+ // catalog until it expires on its own. Clear the whole cache instead;
see the class
+ // javadoc for the classloader cost that comes with it.
+ LOG.error(
+ "Failed to evict catalog {} for change log id {}, clearing the
whole catalog cache to "
+ + "avoid serving it stale; catalogs in use by this node are
closed as a result",
+ ident,
change.getId(),
- change.getFullName(),
- change.getEntityType(),
- change.getOperateType(),
e);
+ catalogManager.getCatalogCache().invalidateAll();
+ return;
}
}
}
+ private static class CatalogInvalidation {
+ private final EntityChangeRecord change;
+ private final NameIdentifier ident;
+
+ private CatalogInvalidation(EntityChangeRecord change, NameIdentifier
ident) {
+ this.change = change;
+ this.ident = ident;
+ }
+ }
+
private boolean isCatalogChange(EntityChangeRecord change) {
if (change.getEntityType() == null) {
return false;
@@ -120,8 +172,11 @@ public class CatalogChangeLogListener implements
EntityChangeLogListener {
NameIdentifier ident;
try {
ident = EntityChangeLogNameIdentifierCodec.decode(change.getFullName());
- } catch (IllegalArgumentException e) {
- LOG.warn("Invalid catalog full name in entity change log: {}",
change.getFullName());
+ } catch (RuntimeException e) {
+ // Catch every unchecked exception, not just IllegalArgumentException:
if a future version of
+ // the codec throws something else, one bad row must still be skipped
instead of aborting the
+ // whole batch, which would drop the invalidations already collected for
the other rows.
+ LOG.warn("Invalid catalog full name in entity change log: {}",
change.getFullName(), e);
return Optional.empty();
}
if (ident.namespace().length() != 1) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
index 587292c6d2..ed7b942e00 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
@@ -52,8 +52,9 @@ import org.slf4j.LoggerFactory;
* <li>A <b>failed invalidation</b> means this node may now serve stale
metadata indefinitely. The
* whole cache is cleared instead, which is strictly stronger than the
invalidation that
* failed and only costs a cold-cache penalty, since the cache is
derived state. If even the
- * clear fails the exception propagates, and {@link
EntityChangeLogPoller} retries the batch
- * and ultimately applies its configured listener failure action.
+ * clear also fails, the exception goes up to {@link
EntityChangeLogPoller}, which only logs
+ * it and moves on. The batch is never sent again, so this node may keep
serving stale entries
+ * until they expire.
* </ul>
*/
public class EntityCacheChangeLogListener implements EntityChangeLogListener {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogCleaner.java
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogCleaner.java
index 29c3ca6bd1..436f0dfce4 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogCleaner.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogCleaner.java
@@ -49,7 +49,7 @@ public class EntityChangeLogCleaner implements AutoCloseable {
/**
* How many poll cycles a change record must survive at minimum. A record
has to outlive more than
* one cycle, because a node can miss cycles while it is restarting, stalled
in a long GC pause,
- * or paused retrying a failed listener.
+ * or slow to drain a large backlog.
*/
private static final long MIN_RETENTION_POLL_CYCLES = 10;
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
index b57d24310b..cbc139e5c3 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogListener.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogListener.java
@@ -28,9 +28,11 @@ public interface EntityChangeLogListener {
/**
* Handles a batch of entity changes.
*
- * <p>If this method throws, the poller may retry the same batch for this
listener.
- * Implementations must make the callback atomic or tolerate retrying
changes that were applied
- * before the exception.
+ * <p>A batch is handed to the listener only once and is never sent again,
so the listener has to
+ * clean up after itself when something goes wrong. The simplest way is to
clear the whole cache
+ * this listener keeps, because that also removes whatever entry it failed
to remove. Do not count
+ * on the poller retrying. If this method throws, the poller only logs the
error at {@code ERROR}
+ * and moves on, and this listener's cache can stay wrong from then on.
*
* @param changes the entity changes fetched in one poller cycle
*/
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
index 34c7e04b1c..afc21cc5b3 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
@@ -20,7 +20,6 @@ package org.apache.gravitino.storage.relational;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
@@ -36,51 +35,44 @@ 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. The cursor advances only after
every listener applies the
- * batch. If a listener throws, the immutable batch stays in memory and the
next poll retries only
- * the listeners that have not succeeded yet.
+ * <p>There is one poller per Gravitino server process, and it keeps one read
position (the id of
+ * the last row it has read). It reads a batch of rows, hands that batch to
every listener once, and
+ * then always moves the read position forward, even if a listener failed. The
read position is
+ * shared by all listeners, so holding it back to retry one listener would
stop every other listener
+ * from seeing new changes, and all the caches in this process would fall
behind.
*
- * <p>A listener that throws after partially applying a batch receives the
whole batch again, so
- * listeners must make each callback atomic or tolerate retrying changes they
already applied. A
- * listener that cannot satisfy that contract must swallow its own failures,
as {@code
- * CatalogChangeLogListener} does.
+ * <p>Because a batch is handed out only once and is never sent again, every
listener must be able
+ * to fix itself when it fails. The usual way to do that is to clear its whole
cache: clearing
+ * everything also covers whatever the listener failed to remove, so nothing
stale is left behind.
+ * The three listeners registered today do exactly that:
*
- * <p>Retries are bounded by {@code maxListenerRetries}. While a batch is
paused no new batch is
- * fetched, so a permanently failing listener would otherwise freeze cache
invalidation for the
- * whole process. When the bound is reached, {@link ListenerFailureAction}
decides what happens:
- * {@code EXIT} stops this server (the local caches are known to be stale, so
serving from them
- * would trade correctness for availability), {@code SKIP} drops the failed
listeners from the batch
- * and advances the cursor.
+ * <ul>
+ * <li>{@code EntityCacheChangeLogListener} clears its whole entity cache.
+ * <li>{@code JcasbinChangeListener} clears its whole {@code
metadataIdCache}. A stale
+ * name→id entry there would be used by authorization checks.
+ * <li>{@code CatalogChangeLogListener} clears its whole catalog cache.
Clearing it also closes
+ * the {@code IsolatedClassLoader} of every cached catalog, including
catalogs this process is
+ * currently serving (that is the bug in #11739). We accept that cost so
a changed catalog is
+ * never served from a stale cache, and the clear only happens when a
normal removal failed.
+ * </ul>
+ *
+ * <p>Do not register a listener here if it cannot recover on its own.
+ *
+ * <p>Every listener failure is logged at {@code ERROR}, so a listener that
keeps failing stays
+ * visible in the logs even though the poller keeps going.
*/
public class EntityChangeLogPoller implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(EntityChangeLogPoller.class);
- /**
- * Max entity-change rows to fetch per batch. A paused batch is retained in
memory until every
- * listener applies it, so this also bounds the poller's retained heap.
- */
+ /** Max entity-change rows to fetch per batch. */
private static final int ENTITY_CHANGE_POLLER_MAX_ROWS = 2000;
/** Max records rendered in a batch summary log line. */
private static final int MAX_SUMMARIZED_RECORDS = 20;
- /** What the poller does when a listener keeps failing after {@code
maxListenerRetries}. */
- public enum ListenerFailureAction {
- /** Stop this server process, because its local caches are known to be
stale. */
- EXIT,
- /** Drop the failing listener from the batch, advance the cursor and keep
serving. */
- SKIP
- }
-
private final List<EntityChangeLogListener> listeners = new
CopyOnWriteArrayList<>();
private final long pollIntervalSecs;
- private final int maxListenerRetries;
- private final ListenerFailureAction listenerFailureAction;
- private final Runnable exitHandler;
-
- @Nullable private BatchDelivery pendingDelivery;
private ScheduledExecutorService scheduler;
private volatile long entityPollHighWaterId = 0;
@@ -89,39 +81,16 @@ public class EntityChangeLogPoller implements AutoCloseable
{
* Creates an {@link EntityChangeLogPoller}.
*
* @param pollIntervalSecs interval between successive polling cycles
- * @param maxListenerRetries how many times a failing listener is retried
for the same batch
- * before {@code listenerFailureAction} is applied
- * @param listenerFailureAction what to do once a listener exhausted its
retries
*/
- public EntityChangeLogPoller(
- long pollIntervalSecs, int maxListenerRetries, ListenerFailureAction
listenerFailureAction) {
- // System.exit() runs the JVM shutdown hooks, which is where
GravitinoServer performs its
- // graceful stop, so in-flight requests still get a chance to finish.
- this(pollIntervalSecs, maxListenerRetries, listenerFailureAction, () ->
System.exit(1));
- }
-
- @VisibleForTesting
- EntityChangeLogPoller(
- long pollIntervalSecs,
- int maxListenerRetries,
- ListenerFailureAction listenerFailureAction,
- Runnable exitHandler) {
+ public EntityChangeLogPoller(long pollIntervalSecs) {
Preconditions.checkArgument(pollIntervalSecs > 0, "pollIntervalSecs must
be positive");
- Preconditions.checkArgument(maxListenerRetries >= 0, "maxListenerRetries
must be non-negative");
- Preconditions.checkArgument(
- listenerFailureAction != null, "listenerFailureAction cannot be null");
this.pollIntervalSecs = pollIntervalSecs;
- this.maxListenerRetries = maxListenerRetries;
- this.listenerFailureAction = listenerFailureAction;
- this.exitHandler = exitHandler;
}
/**
* Registers a listener to receive future entity change batches.
*
- * <p>A listener only receives batches fetched after it was registered. In
particular, if a batch
- * is currently paused by a failing listener, the newly registered listener
does not receive that
- * batch and the cursor moves past it once the batch completes.
+ * <p>A listener only receives batches fetched after it was registered.
*
* @param listener the listener to register
*/
@@ -165,12 +134,10 @@ public class EntityChangeLogPoller implements
AutoCloseable {
EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
LOG.info(
"Starting entity change log poller at high-water id {} with a {}
second interval, "
- + "{} listener(s) registered, maxListenerRetries={},
listenerFailureAction={}",
+ + "{} listener(s) registered",
entityPollHighWaterId,
pollIntervalSecs,
- listeners.size(),
- maxListenerRetries,
- listenerFailureAction);
+ listeners.size());
scheduler =
Executors.newSingleThreadScheduledExecutor(
@@ -200,23 +167,19 @@ public class EntityChangeLogPoller implements
AutoCloseable {
// The final cursor tells where this node stopped consuming, which is the
starting point when
// comparing nodes after an incident.
- LOG.info(
- "Stopped entity change log poller at high-water id {}{}",
- entityPollHighWaterId,
- pendingDelivery == null
- ? ""
- : ", with an unapplied batch id range ["
- + pendingDelivery.firstChangeId()
- + ", "
- + pendingDelivery.lastChangeId
- + "]");
+ LOG.info("Stopped entity change log poller at high-water id {}",
entityPollHighWaterId);
}
@VisibleForTesting
void pollChanges() {
try {
doPollChanges();
- } catch (Exception e) {
+ } catch (Throwable e) {
+ // Catch Throwable, not Exception: this method is the task handed to
+ // scheduleWithFixedDelay(), and anything that escapes it cancels all
future runs for good,
+ // silently. An Error is reachable here, for example a
NoClassDefFoundError thrown by a
+ // listener that touched a closed IsolatedClassLoader. Losing the poller
would stop cache
+ // invalidation for every listener in this process, so we log and let
the next cycle run.
if (handleInterruptIfAny(e, "Entity change poll")) {
return;
}
@@ -225,22 +188,7 @@ public class EntityChangeLogPoller implements
AutoCloseable {
}
private synchronized void doPollChanges() {
- BatchDelivery delivery = pendingDelivery;
- if (delivery != null) {
- LOG.info(
- "Retrying entity change log batch with {} record(s), id range [{},
{}], attempt {} of "
- + "{}, for {} pending listener(s)",
- delivery.changes.size(),
- delivery.firstChangeId(),
- delivery.lastChangeId,
- delivery.attempts,
- maxListenerRetries + 1,
- delivery.pendingListeners.size());
- deliver(delivery);
- return;
- }
-
- delivery = fetchNextDelivery();
+ BatchDelivery delivery = fetchNextDelivery();
if (delivery != null) {
deliver(delivery);
}
@@ -256,7 +204,7 @@ public class EntityChangeLogPoller implements AutoCloseable
{
List<EntityChangeRecord> immutableChanges = List.copyOf(changes);
long lastChangeId = immutableChanges.get(immutableChanges.size() -
1).getId();
BatchDelivery delivery =
- new BatchDelivery(immutableChanges, lastChangeId,
List.copyOf(listeners), 1);
+ new BatchDelivery(immutableChanges, lastChangeId,
List.copyOf(listeners));
LOG.debug(
"Fetched {} entity change log record(s) after cursor {}, id range [{},
{}]: {}",
immutableChanges.size(),
@@ -323,33 +271,13 @@ public class EntityChangeLogPoller implements
AutoCloseable {
}
private void deliver(BatchDelivery delivery) {
- List<EntityChangeLogListener> failedListeners = notifyListeners(delivery);
- if (failedListeners.isEmpty()) {
- advanceCursor(delivery);
- return;
- }
-
- if (delivery.attempts > maxListenerRetries) {
- handleExhaustedRetries(delivery, failedListeners);
- return;
- }
-
- pendingDelivery = delivery.retryOnly(failedListeners);
- LOG.error(
- "Entity change log cursor is paused at id {} because {} listener(s)
failed to apply batch "
- + "id range [{}, {}] (attempt {} of {})",
- entityPollHighWaterId,
- failedListeners.size(),
- delivery.firstChangeId(),
- delivery.lastChangeId,
- delivery.attempts,
- maxListenerRetries + 1);
+ notifyListeners(delivery);
+ advanceCursor(delivery);
}
private void advanceCursor(BatchDelivery delivery) {
long previousHighWaterId = entityPollHighWaterId;
entityPollHighWaterId = delivery.lastChangeId;
- pendingDelivery = null;
LOG.info(
"Consumed {} entity change log record(s), id range [{}, {}]; cursor
advanced from {} to {}; "
+ "newest record is ~{} ms old",
@@ -361,38 +289,13 @@ public class EntityChangeLogPoller implements
AutoCloseable {
delivery.approximateLagMs());
}
- private void handleExhaustedRetries(
- BatchDelivery delivery, List<EntityChangeLogListener> failedListeners) {
- List<String> failedListenerNames = new ArrayList<>();
- for (EntityChangeLogListener listener : failedListeners) {
- failedListenerNames.add(listener.getClass().getName());
- }
-
- if (listenerFailureAction == ListenerFailureAction.EXIT) {
- LOG.error(
- "Stopping this server: listener(s) {} failed to apply entity change
log batch id range "
- + "[{}, {}] after {} attempt(s), so local caches are stale and
cannot be trusted",
- failedListenerNames,
- delivery.firstChangeId(),
- delivery.lastChangeId,
- delivery.attempts);
- exitHandler.run();
- return;
- }
-
- LOG.error(
- "Dropping entity change log batch id range [{}, {}] for listener(s) {}
after {} attempt(s);"
- + " their local caches may be stale until the affected entries
expire",
- delivery.firstChangeId(),
- delivery.lastChangeId,
- failedListenerNames,
- delivery.attempts);
- advanceCursor(delivery);
- }
-
- private List<EntityChangeLogListener> notifyListeners(BatchDelivery
delivery) {
- List<EntityChangeLogListener> failedListeners = new ArrayList<>();
- for (EntityChangeLogListener listener : delivery.pendingListeners) {
+ /**
+ * Hands the batch to every listener that is still registered. If a listener
throws, the error is
+ * only logged: each listener is expected to clean up after itself, and the
read position moves
+ * forward either way.
+ */
+ private void notifyListeners(BatchDelivery delivery) {
+ for (EntityChangeLogListener listener : delivery.targetListeners) {
if (!listeners.contains(listener)) {
LOG.debug(
"Skipping unregistered entity change log listener {} for batch id
range [{}, {}]",
@@ -409,40 +312,33 @@ public class EntityChangeLogPoller implements
AutoCloseable {
listener.getClass().getName(),
delivery.firstChangeId(),
delivery.lastChangeId);
- } catch (Exception e) {
- failedListeners.add(listener);
- LOG.warn(
- "Entity change log listener {} failed to consume batch id range
[{}, {}]",
+ } catch (Throwable e) {
+ // Throwable, not Exception: a listener recovering by clearing its
cache can close an
+ // IsolatedClassLoader that is still in use and surface a
NoClassDefFoundError, which is an
+ // Error. One listener doing that must not take down the whole poller.
+ LOG.error(
+ "Entity change log listener {} failed to consume batch id range
[{}, {}]; the batch is "
+ + "not retried, so the listener is responsible for local
recovery",
listener.getClass().getName(),
delivery.firstChangeId(),
delivery.lastChangeId,
e);
}
}
- return failedListeners;
}
private static class BatchDelivery {
private final List<EntityChangeRecord> changes;
private final long lastChangeId;
- private final List<EntityChangeLogListener> pendingListeners;
-
- /** How many times this batch has been dispatched, starting at 1 for the
initial dispatch. */
- private final int attempts;
+ private final List<EntityChangeLogListener> targetListeners;
private BatchDelivery(
List<EntityChangeRecord> changes,
long lastChangeId,
- List<EntityChangeLogListener> pendingListeners,
- int attempts) {
+ List<EntityChangeLogListener> targetListeners) {
this.changes = changes;
this.lastChangeId = lastChangeId;
- this.pendingListeners = pendingListeners;
- this.attempts = attempts;
- }
-
- private BatchDelivery retryOnly(List<EntityChangeLogListener>
failedListeners) {
- return new BatchDelivery(changes, lastChangeId,
List.copyOf(failedListeners), attempts + 1);
+ this.targetListeners = targetListeners;
}
private long firstChangeId() {
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 6700b83ec1..bdac4f92b8 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
@@ -27,7 +27,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import java.util.Locale;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
@@ -108,13 +107,7 @@ public class RelationalEntityStore
// Polling and cleanup use separate single-threaded schedulers. Polling
only dispatches changes
// to local listeners, while cleanup independently removes records beyond
the retention period.
this.entityChangeLogPoller =
- new EntityChangeLogPoller(
- config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS),
- config.get(Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES),
- EntityChangeLogPoller.ListenerFailureAction.valueOf(
- config
- .get(Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)
- .toUpperCase(Locale.ROOT)));
+ new
EntityChangeLogPoller(config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS));
this.entityChangeLogCleaner =
new EntityChangeLogCleaner(
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)),
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 fbc54b2903..7b6686c6c1 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -138,8 +136,6 @@ public class TestAccessControlManager {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
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 564000a7ca..936d5336a7 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -116,8 +114,6 @@ public class TestOwnerManager {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java
index 4ec22ccd58..056385102f 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java
@@ -19,7 +19,10 @@
package org.apache.gravitino.catalog;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doThrow;
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;
@@ -33,33 +36,148 @@ import
org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
public class TestCatalogChangeLogListener {
@Test
@SuppressWarnings("unchecked")
- void testProcessesRemainingChangesAndSwallowsFailure() {
+ void testFailedClearPropagatesToThePoller() {
CatalogManager catalogManager = mock(CatalogManager.class);
Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
- NameIdentifier failedIdentifier = NameIdentifier.of("metalake", "failed");
- NameIdentifier successfulIdentifier = NameIdentifier.of("metalake",
"successful");
+ NameIdentifier failing = NameIdentifier.of("metalake", "failing");
+ NameIdentifier laterLocal = NameIdentifier.of("metalake", "later_local");
when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
- when(catalogManager.consumeLocalMutation(failedIdentifier))
- .thenThrow(new RuntimeException("invalidation failed"));
+ when(catalogManager.consumeLocalMutation(failing)).thenReturn(false);
+ when(catalogManager.consumeLocalMutation(laterLocal)).thenReturn(true);
+ doThrow(new RuntimeException("eviction
failed")).when(catalogCache).invalidate(failing);
+ doThrow(new RuntimeException("clear
failed")).when(catalogCache).invalidateAll();
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+
+ // There is nothing else we can do here, so let the exception go up. That
is fine now: the
+ // poller only logs it and never sends the batch again. Re-sending was the
dangerous part,
+ // because consumeLocalMutation() works only once and a second pass would
mistake a change made
+ // by this node for one made by another node.
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ listener.onEntityChange(
+ List.of(change(1L, "metalake.failing"), change(2L,
"metalake.later_local"))));
+
+ // The "made by this node" marks are all read before any removal starts,
even when both the
+ // removal and the clear fail.
+ verify(catalogManager).consumeLocalMutation(laterLocal);
+ verify(catalogCache).invalidateAll();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testFailedEvictionClearsTheWholeCatalogCache() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
+ NameIdentifier failing = NameIdentifier.of("metalake", "failing");
+ NameIdentifier laterRemote = NameIdentifier.of("metalake", "later_remote");
+ NameIdentifier laterLocal = NameIdentifier.of("metalake", "later_local");
+
+ when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
+ when(catalogManager.consumeLocalMutation(failing)).thenReturn(false);
+ when(catalogManager.consumeLocalMutation(laterRemote)).thenReturn(false);
+ when(catalogManager.consumeLocalMutation(laterLocal)).thenReturn(true);
+ doThrow(new RuntimeException("eviction
failed")).when(catalogCache).invalidate(failing);
CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
- // The failure must not reach the poller. A retried batch would
re-invalidate catalogs this
- // process mutated itself, because consumeLocalMutation() is single-shot,
and closing an
- // in-use CatalogWrapper also closes its IsolatedClassLoader.
Assertions.assertDoesNotThrow(
() ->
listener.onEntityChange(
- List.of(change(1L, "metalake.failed"), change(2L,
"metalake.successful"))));
+ List.of(
+ change(1L, "metalake.failing"),
+ change(2L, "metalake.later_remote"),
+ change(3L, "metalake.later_local"))));
+
+ // The later local mark is read before any removal starts. Clearing the
whole cache then covers
+ // both the removal that failed and every remote record, so nothing else
has to be removed.
+ verify(catalogManager).consumeLocalMutation(laterLocal);
+ verify(catalogCache).invalidateAll();
+ verify(catalogCache, never()).invalidate(laterRemote);
+ verify(catalogCache, never()).invalidate(laterLocal);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testSuccessfulBatchDoesNotClearTheCatalogCache() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
+ NameIdentifier first = NameIdentifier.of("metalake", "cat1");
+ NameIdentifier second = NameIdentifier.of("metalake", "cat2");
+
+ when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
+ when(catalogManager.consumeLocalMutation(any())).thenReturn(false);
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+
+ listener.onEntityChange(List.of(change(1L, "metalake.cat1"), change(2L,
"metalake.cat2")));
- verify(catalogCache).invalidate(successfulIdentifier);
- verify(catalogCache, never()).invalidate(failedIdentifier);
+ verify(catalogCache).invalidate(first);
+ verify(catalogCache).invalidate(second);
+ // Clearing closes IsolatedClassLoaders that are still in use, so it must
never happen during
+ // normal operation.
+ verify(catalogCache, never()).invalidateAll();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testMalformedRecordIsSkippedWithoutClearingTheCatalogCache() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
+ NameIdentifier healthy = NameIdentifier.of("metalake", "healthy");
+
+ when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
+ when(catalogManager.consumeLocalMutation(any())).thenReturn(false);
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+
+ // A row that does not point at any catalog leaves nothing stale, so it
must not cause a
+ // whole-cache clear.
+ Assertions.assertDoesNotThrow(
+ () ->
+ listener.onEntityChange(
+ List.of(
+ new EntityChangeRecord(1L, "metalake", "CATALOG", null,
OperateType.ALTER, 0L),
+ change(2L, "metalake.cat.schema"),
+ change(3L, "metalake.healthy"))));
+
+ verify(catalogCache).invalidate(healthy);
+ verify(catalogCache, never()).invalidateAll();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testFailedLocalMutationProbeTreatsTheRecordAsRemote() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
+ NameIdentifier failing = NameIdentifier.of("metalake", "failing");
+ NameIdentifier healthy = NameIdentifier.of("metalake", "healthy");
+
+ when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
+ when(catalogManager.consumeLocalMutation(failing))
+ .thenThrow(new RuntimeException("probe failed"));
+ when(catalogManager.consumeLocalMutation(healthy)).thenReturn(false);
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+
+ // When we cannot tell which node made the change, assume another node
did. Removing one entry
+ // for nothing is cheaper than missing a real remote change, which we
would never see again.
+ Assertions.assertDoesNotThrow(
+ () ->
+ listener.onEntityChange(
+ List.of(change(1L, "metalake.failing"), change(2L,
"metalake.healthy"))));
+
+ verify(catalogCache).invalidate(failing);
+ verify(catalogCache).invalidate(healthy);
+ verify(catalogCache, never()).invalidateAll();
}
@Test
@@ -101,6 +219,37 @@ public class TestCatalogChangeLogListener {
verify(catalogCache).invalidate(ident);
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void testUnexpectedDecodeFailureSkipsOnlyThatRecord() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
+ NameIdentifier healthy = NameIdentifier.of("metalake", "healthy");
+
+ when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
+ when(catalogManager.consumeLocalMutation(any())).thenReturn(false);
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+
+ // The codec throws IllegalArgumentException today, but that is an
implementation detail two
+ // calls down. If it ever throws something else, one bad row must still be
skipped instead of
+ // aborting the batch and dropping the invalidations already collected for
the other rows.
+ try (MockedStatic<EntityChangeLogNameIdentifierCodec> codec =
+ mockStatic(EntityChangeLogNameIdentifierCodec.class,
CALLS_REAL_METHODS)) {
+ codec
+ .when(() ->
EntityChangeLogNameIdentifierCodec.decode("metalake.boom"))
+ .thenThrow(new IllegalStateException("codec blew up"));
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ listener.onEntityChange(
+ List.of(change(1L, "metalake.boom"), change(2L,
"metalake.healthy"))));
+ }
+
+ verify(catalogCache).invalidate(healthy);
+ verify(catalogCache, never()).invalidateAll();
+ }
+
private static EntityChangeRecord change(long id, String fullName) {
return new EntityChangeRecord(id, "metalake", "CATALOG", fullName,
OperateType.ALTER, 0L);
}
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 fe0447fa75..76daebb873 100644
---
a/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -221,8 +219,6 @@ public class TestFilesetHookDispatcher extends
TestOperationDispatcher {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
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 46504db01c..5b80249042 100644
--- a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
+++ b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -201,8 +199,6 @@ public class TestPolicyManager {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
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 a8a7538ba9..aa0cb0fea5 100644
--- a/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
+++ b/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -115,8 +113,6 @@ public class TestStatisticManager {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
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 31e39e3388..cac53eed61 100644
---
a/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
+++
b/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
@@ -21,8 +21,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -149,8 +147,6 @@ abstract class AbstractEntityStorageTest {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
index 9f75ec105d..5c979bdfb2 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
@@ -45,12 +45,7 @@ public class TestEntityCacheCrossNodeInvalidation extends
TestJDBCBackend {
// A large poll interval so the background scheduler never fires during the
test; the test drives
// node B's poll explicitly via pollChanges().
private EntityChangeLogPoller newIdlePoller(CaffeineEntityCache nodeBCache) {
- EntityChangeLogPoller poller =
- new EntityChangeLogPoller(
- 3600,
- 0,
- EntityChangeLogPoller.ListenerFailureAction.EXIT,
- () -> Assertions.fail("The entity cache listener must not exhaust
its retries"));
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(3600);
poller.registerListener(new EntityCacheChangeLogListener(nodeBCache));
// start() seeds the cursor with the current DB tail, modelling a node
whose cache is already
// warm: only changes written after this point are replayed.
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
index bf54d5b6ef..37c05bd18b 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java
@@ -19,21 +19,15 @@
package org.apache.gravitino.storage.relational;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
-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.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
-import
org.apache.gravitino.storage.relational.EntityChangeLogPoller.ListenerFailureAction;
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;
@@ -48,11 +42,8 @@ public class TestEntityChangeLogPoller {
@Test
void testRejectsInvalidConfiguration() {
- Assertions.assertThrows(IllegalArgumentException.class, () -> newPoller(0,
10));
- Assertions.assertThrows(IllegalArgumentException.class, () ->
newPoller(-1, 10));
- Assertions.assertThrows(IllegalArgumentException.class, () -> newPoller(1,
-1));
- Assertions.assertThrows(
- IllegalArgumentException.class, () -> new EntityChangeLogPoller(1, 10,
null));
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(0));
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(-1));
}
@Test
@@ -69,7 +60,7 @@ public class TestEntityChangeLogPoller {
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller = newPoller(1, 10);
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
poller.registerListener(firstListenerRecords::addAll);
poller.registerListener(secondListenerRecords::addAll);
@@ -83,26 +74,25 @@ public class TestEntityChangeLogPoller {
}
@Test
- void testRetriesOnlyFailedListenersBeforeAdvancingCursor() {
+ void testThrowingListenerNeitherPausesCursorNorBlocksOtherListeners() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
- when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
- when(mapper.selectEntityChanges(1L, MAX_ROWS)).thenReturn(List.of());
+ EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
+ EntityChangeRecord second = change(2L, "CATALOG", "ml1.cat2");
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(first));
+ when(mapper.selectEntityChanges(1L, MAX_ROWS)).thenReturn(List.of(second));
+ when(mapper.selectEntityChanges(2L, MAX_ROWS)).thenReturn(List.of());
+ AtomicInteger throwingListenerCalls = new AtomicInteger();
List<EntityChangeRecord> received = new ArrayList<>();
- AtomicInteger failingListenerCalls = new AtomicInteger();
- AtomicBoolean firstCall = new AtomicBoolean(true);
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller = newPoller(1, 10);
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
poller.registerListener(
changes -> {
- failingListenerCalls.incrementAndGet();
- if (firstCall.getAndSet(false)) {
- throw new RuntimeException("listener failed");
- }
+ throwingListenerCalls.incrementAndGet();
+ throw new RuntimeException("listener failed");
});
poller.registerListener(received::addAll);
@@ -111,127 +101,73 @@ public class TestEntityChangeLogPoller {
poller.pollChanges();
}
- // The healthy listener is not re-notified, the failed one is retried once
and then succeeds.
- Assertions.assertEquals(2, failingListenerCalls.get());
- Assertions.assertEquals(List.of(change), received);
- verify(mapper).selectEntityChanges(0L, MAX_ROWS);
- verify(mapper).selectEntityChanges(1L, MAX_ROWS);
- }
-
- @Test
- void testUnregisteredFailedListenerDoesNotBlockCursor() {
- EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
- when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
- when(mapper.selectEntityChanges(1L, MAX_ROWS)).thenReturn(List.of());
-
- try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
- mockSessionUtils(sessionUtils, mapper);
-
- EntityChangeLogPoller poller = newPoller(1, 10);
- EntityChangeLogListener failedListener =
- changes -> {
- throw new RuntimeException("listener failed");
- };
- poller.registerListener(failedListener);
-
- poller.pollChanges();
- poller.unregisterListener(failedListener);
- poller.pollChanges();
- poller.pollChanges();
- }
-
+ // Each batch is handed out exactly once: the failing listener never gets
a batch a second
+ // time, the healthy listener still gets every batch, and the read
position moves past both.
+ Assertions.assertEquals(2, throwingListenerCalls.get());
+ Assertions.assertEquals(List.of(first, second), received);
verify(mapper).selectEntityChanges(1L, MAX_ROWS);
+ verify(mapper).selectEntityChanges(2L, MAX_ROWS);
}
@Test
- void testPausedBatchBlocksFetchingNewBatches() {
- EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
- when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
-
- try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
- mockSessionUtils(sessionUtils, mapper);
-
- EntityChangeLogPoller poller = newPoller(1, 10);
- poller.registerListener(
- changes -> {
- throw new RuntimeException("listener failed");
- });
-
- poller.pollChanges();
- poller.pollChanges();
- }
-
- // Only the very first fetch happened: the second poll retried the paused
batch instead.
- verify(mapper).selectEntityChanges(0L, MAX_ROWS);
- verify(mapper, times(1)).selectEntityChanges(anyLong(), anyInt());
- }
-
- @Test
- void testStopsServerAfterListenerExhaustsRetries() {
+ void testListenerThrowingErrorDoesNotKillThePoller() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
- when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
+ EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
+ EntityChangeRecord second = change(2L, "CATALOG", "ml1.cat2");
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(first));
+ when(mapper.selectEntityChanges(1L, MAX_ROWS)).thenReturn(List.of(second));
+ when(mapper.selectEntityChanges(2L, MAX_ROWS)).thenReturn(List.of());
- AtomicInteger listenerCalls = new AtomicInteger();
- AtomicInteger exitCalls = new AtomicInteger();
+ List<EntityChangeRecord> received = new ArrayList<>();
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller =
- new EntityChangeLogPoller(1, 1, ListenerFailureAction.EXIT,
exitCalls::incrementAndGet);
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ // A listener that clears its catalog cache can close an
IsolatedClassLoader that is still in
+ // use, and a request holding a class from it then fails with
NoClassDefFoundError, an Error
+ // rather than an Exception. pollChanges() is the task given to
scheduleWithFixedDelay(), so
+ // anything escaping it would cancel every future poll and freeze
invalidation process-wide.
poller.registerListener(
changes -> {
- listenerCalls.incrementAndGet();
- throw new RuntimeException("listener failed");
+ throw new NoClassDefFoundError("closed isolated classloader");
});
+ poller.registerListener(received::addAll);
- poller.pollChanges();
- Assertions.assertEquals(0, exitCalls.get());
- poller.pollChanges();
+ Assertions.assertDoesNotThrow(poller::pollChanges);
+ Assertions.assertDoesNotThrow(poller::pollChanges);
+ Assertions.assertDoesNotThrow(poller::pollChanges);
}
- // One initial dispatch plus one retry, then the server is stopped and the
cursor never moves.
- Assertions.assertEquals(2, listenerCalls.get());
- Assertions.assertEquals(1, exitCalls.get());
- verify(mapper, never()).selectEntityChanges(1L, MAX_ROWS);
+ Assertions.assertEquals(List.of(first, second), received);
+ verify(mapper).selectEntityChanges(1L, MAX_ROWS);
+ verify(mapper).selectEntityChanges(2L, MAX_ROWS);
}
@Test
- void testSkipActionDropsFailedListenerAndAdvancesCursor() {
+ void testUnregisteredListenerIsSkipped() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
EntityChangeRecord second = change(2L, "CATALOG", "ml1.cat2");
when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(first));
when(mapper.selectEntityChanges(1L, MAX_ROWS)).thenReturn(List.of(second));
- when(mapper.selectEntityChanges(2L, MAX_ROWS)).thenReturn(List.of());
- AtomicInteger listenerCalls = new AtomicInteger();
+ List<EntityChangeRecord> received = new ArrayList<>();
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller =
- new EntityChangeLogPoller(
- 1, 0, ListenerFailureAction.SKIP,
TestEntityChangeLogPoller::failOnExit);
- poller.registerListener(
- changes -> {
- listenerCalls.incrementAndGet();
- throw new RuntimeException("listener failed");
- });
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ EntityChangeLogListener listener = received::addAll;
+ poller.registerListener(listener);
poller.pollChanges();
- poller.pollChanges();
+ poller.unregisterListener(listener);
poller.pollChanges();
}
- // No retry with maxListenerRetries=0: each batch is dropped for the
listener after one
- // attempt, and the cursor keeps moving.
- Assertions.assertEquals(2, listenerCalls.get());
+ Assertions.assertEquals(List.of(first), received);
verify(mapper).selectEntityChanges(1L, MAX_ROWS);
- verify(mapper).selectEntityChanges(2L, MAX_ROWS);
}
@Test
@@ -246,7 +182,7 @@ public class TestEntityChangeLogPoller {
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller = newPoller(1, 10);
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
poller.registerListener(received::addAll);
poller.pollChanges();
@@ -265,7 +201,7 @@ public class TestEntityChangeLogPoller {
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller = newPoller(1, 10);
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
Assertions.assertDoesNotThrow(poller::pollChanges);
}
@@ -284,7 +220,7 @@ public class TestEntityChangeLogPoller {
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller = newPoller(1, 10);
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
poller.registerListener(
changes ->
Assertions.assertThrows(UnsupportedOperationException.class, changes::clear));
poller.registerListener(received::addAll);
@@ -295,19 +231,6 @@ public class TestEntityChangeLogPoller {
Assertions.assertEquals(List.of(first, second), received);
}
- /** Fails the test instead of stopping the JVM when the poller decides to
exit. */
- private static void failOnExit() {
- Assertions.fail("the poller must not stop the server in this scenario");
- }
-
- private static EntityChangeLogPoller newPoller(long pollIntervalSecs, int
maxListenerRetries) {
- return new EntityChangeLogPoller(
- pollIntervalSecs,
- maxListenerRetries,
- ListenerFailureAction.EXIT,
- TestEntityChangeLogPoller::failOnExit);
- }
-
private static EntityChangeRecord change(long id, String type, String
fullName) {
return new EntityChangeRecord(id, "ml1", type, fullName,
OperateType.ALTER, 0L);
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
index bb1bfbd3e9..77a8fd419f 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
@@ -162,8 +162,6 @@ public class TestRelationalEntityStoreHierarchicalCache {
Mockito.when(config.get(Configs.STORE_DELETE_AFTER_TIME)).thenReturn(20 *
60 * 1000L);
Mockito.when(config.get(Configs.VERSION_RETENTION_COUNT)).thenReturn(1L);
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
-
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24
* 60 * 60L);
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
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 a683561823..e19008836f 100644
--- a/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
+++ b/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
@@ -20,8 +20,6 @@ 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_LISTENER_FAILURE_ACTION;
-import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
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;
@@ -157,8 +155,6 @@ public class TestTagManager {
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_LISTENER_MAX_RETRIES)).thenReturn(10);
-
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
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);
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index 2a23251322..c2ad57ba33 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -307,13 +307,11 @@ Caches are local to each server, so a metalake modified
on one server would othe
its neighbors. Every server writes its changes to an entity change log table
and polls that table
to invalidate what other servers have touched. A separate cleaner trims old
rows.
-| Configuration Item | Description
| Default Value
|
-|---------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------|
-| `gravitino.entityChangeLog.pollIntervalSecs` | Interval in seconds
between polls. Must be positive.
| `3`
|
-| `gravitino.entityChangeLog.listenerMaxRetries` | Times a batch is retried
for a failing listener before `listenerFailureAction` applies. Must be
non-negative. | `10`
|
-| `gravitino.entityChangeLog.listenerFailureAction` | What happens when a
listener exhausts its retries. `EXIT` stops the server, on the grounds that its
caches are known stale; `SKIP` drops the batch and keeps serving. | `EXIT`
|
-| `gravitino.entityChangeLog.retentionSecs` | How long in seconds
change log rows are kept, measured by database time. `0` disables cleanup;
otherwise use at least ten times `pollIntervalSecs`. |
`2592000` (30 days) |
-| `gravitino.entityChangeLog.cleanupIntervalSecs` | Interval in seconds
between cleaner runs. Must be positive.
| `86400` (1
day) |
+| Configuration Item | Description
| Default Value |
+|-------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|---------------------|
+| `gravitino.entityChangeLog.pollIntervalSecs` | Interval in seconds
between polls. Must be positive.
| `3` |
+| `gravitino.entityChangeLog.retentionSecs` | How long in seconds change
log rows are kept, measured by database time. `0` disables cleanup; otherwise
use at least ten times `pollIntervalSecs`. | `2592000` (30 days) |
+| `gravitino.entityChangeLog.cleanupIntervalSecs` | Interval in seconds
between cleaner runs. Must be positive.
| `86400` (1 day) |
#### Tree Lock
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java
index 06dcf80f1b..7ef37273e0 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java
@@ -225,10 +225,16 @@ public class JcasbinChangeListener implements
EntityChangeLogListener, AutoClose
* change starts emitting the new post-rename name, this invalidation will
silently miss and stale
* entries will only clear via LRU eviction.
*
- * <p><b>Poison-row tolerance:</b> a record this listener cannot map is
logged and skipped rather
- * than propagated. The poller treats a listener throw as a batch failure
and, after exhausting
- * retries, its default action is to stop the server — so one unmappable row
must never be able to
- * take the node down.
+ * <p><b>Bad rows:</b> a record this listener cannot understand is logged
and skipped instead of
+ * being thrown up. Such a row does not point at any cache key, so skipping
it leaves nothing
+ * stale behind.
+ *
+ * <p><b>Recovering from a failure:</b> the poller hands each batch over
only once and never sends
+ * it again, so a failed removal has to be handled here. Otherwise {@code
metadataIdCache} would
+ * keep an out-of-date name→id entry and hand it to authorization
checks until that entry's
+ * TTL runs out. So the whole {@code metadataIdCache} is cleared instead.
That is safe: everything
+ * in it can be looked up again, and clearing it also covers the entry that
failed plus the rest
+ * of the batch. The only cost is redoing those name→id lookups.
*
* <p>The {@code synchronized} modifier is defensive — see the note on
{@link #pollOwnerChanges()}
* for the rationale. The single-threaded scheduler already prevents
overlapping runs in
@@ -325,18 +331,44 @@ public class JcasbinChangeListener implements
EntityChangeLogListener, AutoClose
if (prefixes.isEmpty() && leafKeys.isEmpty()) {
return;
}
- // Hold the cache's exclusive invalidation lock for the whole batch so
readers never observe
- // a half-applied state where some prefix/leaf keys have been evicted and
others have not.
- metadataIdCache.runInvalidationBatch(
- () -> {
- for (String prefix : prefixes) {
- metadataIdCache.invalidateByPrefix(prefix);
- }
- for (String leafKey : leafKeys) {
- if (prefixes.stream().noneMatch(leafKey::startsWith)) {
- metadataIdCache.invalidate(leafKey);
+ // Take the cache's exclusive lock for the whole batch, so a reader never
sees a state where
+ // part of the batch has been removed and part has not. If removing one
key fails, clear the
+ // whole cache while we still hold the lock. If we never got the lock at
all, clear without it.
+ boolean[] batchStarted = {false};
+ try {
+ metadataIdCache.runInvalidationBatch(
+ () -> {
+ batchStarted[0] = true;
+ try {
+ for (String prefix : prefixes) {
+ metadataIdCache.invalidateByPrefix(prefix);
+ }
+ for (String leafKey : leafKeys) {
+ if (prefixes.stream().noneMatch(leafKey::startsWith)) {
+ metadataIdCache.invalidate(leafKey);
+ }
+ }
+ } catch (RuntimeException e) {
+ LOG.error(
+ "Failed to invalidate {} prefix(es) and {} leaf key(s) from
the entity change "
+ + "log, clearing the whole metadata id cache to stay
coherent",
+ prefixes.size(),
+ leafKeys.size(),
+ e);
+ metadataIdCache.invalidateAll();
}
- }
- });
+ });
+ } catch (RuntimeException e) {
+ if (batchStarted[0]) {
+ throw e;
+ }
+ LOG.error(
+ "Failed to start an invalidation batch for {} prefix(es) and {} leaf
key(s), clearing "
+ + "the whole metadata id cache without the batch lock",
+ prefixes.size(),
+ leafKeys.size(),
+ e);
+ metadataIdCache.invalidateAll();
+ }
}
}
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 2315405c63..8204892c4f 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
@@ -150,8 +150,8 @@ public class TestJcasbinChangePoller {
JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
- // A record whose full name cannot be turned into a MetadataObject of the
declared type must be
- // logged and skipped; a poison row must never take the node down via the
poller's EXIT action.
+ // If the full name does not match the type in the record, we cannot build
a cache key from it.
+ // Log it, skip it, and keep handling the rest of the batch.
Assertions.assertDoesNotThrow(
() ->
poller.onEntityChange(
@@ -166,6 +166,105 @@ public class TestJcasbinChangePoller {
Assertions.assertEquals(List.of(), metadataIdCache.invalidatedKeys);
}
+ @Test
+ void testLeafTypesAreInvalidatedByExactKey() {
+ RecordingCache<String, Long> metadataIdCache = new RecordingCache<>();
+ RecordingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
RecordingCache<>();
+
+ JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
+ poller.onEntityChange(List.of(change(1L, MetadataObject.Type.FILESET,
"ml1.cat1.sch1.fs1")));
+
+ // A FILESET has nothing nested under it, so it is removed by its exact
key, not by prefix.
+ Assertions.assertEquals(
+ List.of(key("ml1", "CATALOG", "cat1", "SCHEMA", "sch1", "FILESET",
"fs1")),
+ metadataIdCache.invalidatedKeys);
+ Assertions.assertEquals(List.of(), metadataIdCache.invalidatedPrefixes);
+ Assertions.assertEquals(0, metadataIdCache.invalidateAllCalls);
+ }
+
+ @Test
+ void testSuccessfulBatchDoesNotClearTheCache() {
+ RecordingCache<String, Long> metadataIdCache = new RecordingCache<>();
+ RecordingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
RecordingCache<>();
+
+ JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
+ poller.onEntityChange(
+ List.of(
+ change(1L, MetadataObject.Type.CATALOG, "ml1.cat1"),
+ change(2L, MetadataObject.Type.FILESET, "ml1.cat2.sch1.fs1")));
+
+ Assertions.assertEquals(0, metadataIdCache.invalidateAllCalls);
+ Assertions.assertEquals(0, ownerRelCache.invalidateAllCalls);
+ }
+
+ @Test
+ void testFailedPrefixInvalidationClearsTheWholeMetadataIdCache() {
+ RecordingCache<String, Long> metadataIdCache = new RecordingCache<>();
+ metadataIdCache.failPrefixInvalidation = true;
+ RecordingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
RecordingCache<>();
+
+ JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
+
+ // The batch is handed out once and never sent again, so the listener has
to fix things here.
+ Assertions.assertDoesNotThrow(
+ () -> poller.onEntityChange(List.of(change(1L,
MetadataObject.Type.CATALOG, "ml1.cat1"))));
+
+ Assertions.assertEquals(1, metadataIdCache.invalidateAllCalls);
+ Assertions.assertEquals(1, metadataIdCache.invalidateAllInsideBatchCalls);
+ // The owner cache has its own poller, so a change-log failure must not
wipe it as well.
+ Assertions.assertEquals(0, ownerRelCache.invalidateAllCalls);
+ }
+
+ @Test
+ void testFailedLeafInvalidationClearsTheWholeMetadataIdCache() {
+ RecordingCache<String, Long> metadataIdCache = new RecordingCache<>();
+ metadataIdCache.failKeyInvalidation = true;
+ RecordingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
RecordingCache<>();
+
+ JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ poller.onEntityChange(
+ List.of(change(1L, MetadataObject.Type.FILESET,
"ml1.cat1.sch1.fs1"))));
+
+ Assertions.assertEquals(1, metadataIdCache.invalidateAllCalls);
+ Assertions.assertEquals(1, metadataIdCache.invalidateAllInsideBatchCalls);
+ }
+
+ @Test
+ void testFailedInvalidationBatchClearsTheWholeMetadataIdCache() {
+ RecordingCache<String, Long> metadataIdCache = new RecordingCache<>();
+ metadataIdCache.failInvalidationBatch = true;
+ RecordingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
RecordingCache<>();
+
+ JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
+
+ // The failure can also happen while taking the cache's batch lock, before
any key is touched.
+ Assertions.assertDoesNotThrow(
+ () -> poller.onEntityChange(List.of(change(1L,
MetadataObject.Type.CATALOG, "ml1.cat1"))));
+
+ Assertions.assertEquals(1, metadataIdCache.invalidateAllCalls);
+ Assertions.assertEquals(0, metadataIdCache.invalidateAllInsideBatchCalls);
+ }
+
+ @Test
+ void testFailedClearPropagatesToThePoller() {
+ RecordingCache<String, Long> metadataIdCache = new RecordingCache<>();
+ metadataIdCache.failPrefixInvalidation = true;
+ metadataIdCache.failInvalidateAll = true;
+ RecordingCache<Long, Optional<OwnerInfo>> ownerRelCache = new
RecordingCache<>();
+
+ JcasbinChangeListener poller = new JcasbinChangeListener(metadataIdCache,
ownerRelCache, 1);
+
+ // There is nothing else to try here, so the poller just logs it and keeps
reading.
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> poller.onEntityChange(List.of(change(1L,
MetadataObject.Type.CATALOG, "ml1.cat1"))));
+
+ Assertions.assertEquals(1, metadataIdCache.invalidateAllCalls);
+ }
+
@Test
void testPollCursorAdvancementIsSynchronized() throws NoSuchMethodException {
Method pollOwnerChanges =
JcasbinChangeListener.class.getDeclaredMethod("pollOwnerChanges");
@@ -188,6 +287,14 @@ public class TestJcasbinChangePoller {
private final List<K> invalidatedKeys = new ArrayList<>();
private final List<String> invalidatedPrefixes = new ArrayList<>();
+ private int invalidateAllCalls;
+ private int invalidateAllInsideBatchCalls;
+ private boolean invalidationBatchActive;
+ private boolean failKeyInvalidation;
+ private boolean failPrefixInvalidation;
+ private boolean failInvalidationBatch;
+ private boolean failInvalidateAll;
+
@Override
public Optional<V> getIfPresent(K key) {
return Optional.empty();
@@ -198,17 +305,44 @@ public class TestJcasbinChangePoller {
@Override
public void invalidate(K key) {
+ if (failKeyInvalidation) {
+ throw new RuntimeException("invalidate failed");
+ }
invalidatedKeys.add(key);
}
@Override
- public void invalidateAll() {}
+ public void invalidateAll() {
+ invalidateAllCalls++;
+ if (invalidationBatchActive) {
+ invalidateAllInsideBatchCalls++;
+ }
+ if (failInvalidateAll) {
+ throw new RuntimeException("invalidateAll failed");
+ }
+ }
@Override
public void invalidateByPrefix(String prefix) {
+ if (failPrefixInvalidation) {
+ throw new RuntimeException("invalidateByPrefix failed");
+ }
invalidatedPrefixes.add(prefix);
}
+ @Override
+ public void runInvalidationBatch(Runnable batch) {
+ if (failInvalidationBatch) {
+ throw new RuntimeException("invalidation batch failed");
+ }
+ invalidationBatchActive = true;
+ try {
+ batch.run();
+ } finally {
+ invalidationBatchActive = false;
+ }
+ }
+
@Override
public long size() {
return 0;
