[#12440] improvement(core): Replace the entity change log listener retry/EXIT policy with a cache-clear fallback

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

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

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);

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

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);

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<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")

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 @@
  * <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 CatalogChangeLogListener(CatalogManager catalogManager) {
 
   @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 @@ private Optional<NameIdentifier> 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) {

core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java

@@ -52,8 +52,9 @@
  *   <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 {

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;
 

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