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 d8a4f77c130..fe610e7f0f8 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 @@ 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 static void setUp() throws IllegalAccessException { 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 dd30df96f3e..16b49512361 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 @@ 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 static void setUp() throws IllegalAccessException { 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 c1baa184bf6..fe749a18a3c 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 @@ 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 static void setUp() throws IOException, IllegalAccessException { 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 a95892b088b..3237e961059 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 @@ 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 static void setUp() throws IOException, IllegalAccessException { 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 3fa058e3b52..ba733f5f160 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 @@ 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 @@ protected void initStore() throws IOException { 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 fb1b0a0c7c6..4ed70e1bec6 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 @@ 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.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 @@ private 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 @@ private Configs() {} .checkValue(value -> value > 0, ConfigConstants.POSITIVE_NUMBER_ERROR_MSG) .createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS); - public static final ConfigEntry 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 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 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 9aa3c843c6e..38d9807a0ea 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 @@ *

This listener is called synchronously in the poller thread. Implementations must not * block or perform expensive I/O; only fast, in-memory cache invalidations are permitted. * - *

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}. + *

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. + * + *

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. + * + *

What clearing costs: 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 CatalogChangeLogListener(CatalogManager catalogManager) { @Override public void onEntityChange(List changes) { + List remoteInvalidations = new ArrayList<>(); for (EntityChangeRecord change : changes) { + if (!isCatalogChange(change)) { + continue; + } + + Optional 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 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 @@ private Optional catalogIdentifier(EntityChangeRecord change) { 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 587292c6d2e..ed7b942e00a 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 @@ *

  • A failed invalidation 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. * */ 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 29c3ca6bd17..436f0dfce44 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 b57d24310be..cbc139e5c3e 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. * - *

    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. + *

    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 34c7e04b1cd..afc21cc5b30 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 @@ 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 @@ /** * Global poller for {@code entity_change_log}. * - *

    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. + *

    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. * - *

    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. + *

    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: * - *

    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. + *

      + *
    • {@code EntityCacheChangeLogListener} clears its whole entity cache. + *
    • {@code JcasbinChangeListener} clears its whole {@code metadataIdCache}. A stale + * name→id entry there would be used by authorization checks. + *
    • {@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. + *
    + * + *

    Do not register a listener here if it cannot recover on its own. + * + *

    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 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 enum ListenerFailureAction { * 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. * - *

    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. + *

    A listener only receives batches fetched after it was registered. * * @param listener the listener to register */ @@ -165,12 +134,10 @@ public void start() { 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 void close() { // 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 @@ void pollChanges() { } 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 @@ private BatchDelivery fetchNextDelivery() { List 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 @@ private static long getOrDefault(Long value) { } private void deliver(BatchDelivery delivery) { - List 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 @@ private void advanceCursor(BatchDelivery delivery) { delivery.approximateLagMs()); } - private void handleExhaustedRetries( - BatchDelivery delivery, List failedListeners) { - List 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 notifyListeners(BatchDelivery delivery) { - List 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 @@ private List notifyListeners(BatchDelivery delivery) { 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 changes; private final long lastChangeId; - private final List pendingListeners; - - /** How many times this batch has been dispatched, starting at 1 for the initial dispatch. */ - private final int attempts; + private final List targetListeners; private BatchDelivery( List changes, long lastChangeId, - List pendingListeners, - int attempts) { + List targetListeners) { this.changes = changes; this.lastChangeId = lastChangeId; - this.pendingListeners = pendingListeners; - this.attempts = attempts; - } - - private BatchDelivery retryOnly(List 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 6700b83ec17..bdac4f92b84 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.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 void initialize(Config config) throws RuntimeException { // 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 fbc54b29030..7b6686c6c1b 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 @@ 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 static void setUp() throws Exception { 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 564000a7ca2..936d5336a79 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 @@ 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 static void setUp() throws IOException, IllegalAccessException { 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 4ec22ccd588..056385102f2 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.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 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 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 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 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 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 @@ void testInvalidatesCatalogWithDotsInsideNameLevels() { verify(catalogCache).invalidate(ident); } + @Test + @SuppressWarnings("unchecked") + void testUnexpectedDecodeFailureSkipsOnlyThatRecord() { + CatalogManager catalogManager = mock(CatalogManager.class); + Cache 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 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 fe0447fa750..76daebb8730 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 @@ 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 void testDropAuthorizationPrivilege() { 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 46504db01c7..5b80249042b 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 @@ 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 @@ private static Config mockConfig() { 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 a8a7538ba98..aa0cb0fea58 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 @@ 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 static void setUp() throws IOException, IllegalAccessException { 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 31e39e3388c..cac53eed61c 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 @@ 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 @@ protected void init(String type, Config config) throws IllegalAccessException { 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 9f75ec105d3..5c979bdfb26 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 bf54d5b6ef4..37c05bd18b7 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 @@ void testPollChangesDispatchesSameBatchToAllListenersAndAdvancesCursor() { try (MockedStatic 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 @@ void testPollChangesDispatchesSameBatchToAllListenersAndAdvancesCursor() { } @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 received = new ArrayList<>(); - AtomicInteger failingListenerCalls = new AtomicInteger(); - AtomicBoolean firstCall = new AtomicBoolean(true); try (MockedStatic 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 @@ void testRetriesOnlyFailedListenersBeforeAdvancingCursor() { 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 = 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 = 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 received = new ArrayList<>(); try (MockedStatic 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 received = new ArrayList<>(); try (MockedStatic 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 @@ void testConsumesBacklogLargerThanOneBatch() { try (MockedStatic 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 @@ void testPollChangesCatchesFetchFailures() { try (MockedStatic 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 @@ void testDispatchesImmutableBatchToListeners() { try (MockedStatic 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 @@ void testDispatchesImmutableBatchToListeners() { 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 bb1bfbd3e99..77a8fd419f0 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 @@ private void initStore(String separator) throws Exception { 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 a6835618231..e19008836fa 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 @@ 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 static void setUp() throws IOException, IllegalAccessException { 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 2a23251322a..c2ad57ba337 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 06dcf80f1b8..7ef37273e04 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 @@ private synchronized void pollOwnerChanges() { * change starts emitting the new post-rename name, this invalidation will silently miss and stale * entries will only clear via LRU eviction. * - *

    Poison-row tolerance: 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. + *

    Bad rows: 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. + * + *

    Recovering from a failure: 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. * *

    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 @@ private void invalidateCoalescedKeys(Set prefixes, Set leafKeys) 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 2315405c636..8204892c4f4 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 @@ void testUnmappableRecordDoesNotFailTheBatch() { 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 @@ void testUnmappableRecordDoesNotFailTheBatch() { Assertions.assertEquals(List.of(), metadataIdCache.invalidatedKeys); } + @Test + void testLeafTypesAreInvalidatedByExactKey() { + RecordingCache metadataIdCache = new RecordingCache<>(); + RecordingCache> 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 metadataIdCache = new RecordingCache<>(); + RecordingCache> 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 metadataIdCache = new RecordingCache<>(); + metadataIdCache.failPrefixInvalidation = true; + RecordingCache> 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 metadataIdCache = new RecordingCache<>(); + metadataIdCache.failKeyInvalidation = true; + RecordingCache> 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 metadataIdCache = new RecordingCache<>(); + metadataIdCache.failInvalidationBatch = true; + RecordingCache> 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 metadataIdCache = new RecordingCache<>(); + metadataIdCache.failPrefixInvalidation = true; + metadataIdCache.failInvalidateAll = true; + RecordingCache> 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 @@ private static class RecordingCache implements GravitinoCache { private final List invalidatedKeys = new ArrayList<>(); private final List 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 getIfPresent(K key) { return Optional.empty(); @@ -198,17 +305,44 @@ public void put(K key, V value) {} @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;