[#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,24 @@
  * <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 requires each listener to be self-healing, and this one recovers the same way
+ * {@code EntityCacheChangeLogListener} and {@code JcasbinChangeListener} do: a failed eviction
+ * clears the whole catalog cache, which is a strict superset of the eviction that failed and of the
+ * rest of the batch. A malformed row is skipped instead, because it names no catalog and so leaves
+ * nothing stale.
+ *
+ * <p>The listener consumes local-mutation markers for the whole batch before evicting anything. A
+ * failed eviction or clear therefore cannot strand a marker that would make a later remote change
+ * look local. If the clear itself fails, the exception reaches the poller, which logs it at {@code
+ * ERROR} and advances its cursor; the affected catalog can then remain stale until it expires.
+ *
+ * <p><b>Cost of the clear:</b> evicting a cached catalog closes its {@code CatalogWrapper}, which
+ * tears down the connection pool and the {@code IsolatedClassLoader}. A whole-cache clear therefore
+ * also closes catalogs this process is actively serving; requests holding classes from a closed
+ * loader can fail with {@code NoClassDefFoundError}, the failure mode of #11739. This is accepted
+ * deliberately so that a stale catalog is never served: the alternative left this node serving the
+ * changed catalog from cache for up to {@code gravitino.catalog.cache.evictionIntervalMs}. The
+ * clear runs only on a failed eviction, which is off the normal path.
  */
 public class CatalogChangeLogListener implements EntityChangeLogListener {
 
@@ -59,51 +72,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. A row that names no catalog cannot leave a stale entry behind, so it is
+        // skipped rather than escalated to a cache clear.
+        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) {
+        // The identifier is valid, so this record may name a remote mutation. Treating an unknown
+        // origin as remote can cause an unnecessary eviction, but skipping it can leave a stale
+        // catalog cached after the poller advances its cursor.
+        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;
+      // 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 {}",
+          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={}",
+        // The poller dispatches a batch once and never replays it, so dropping this eviction would
+        // serve the catalog stale until the eviction interval expires. The whole cache is cleared
+        // instead; see the class javadoc for the classloader cost this accepts.
+        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;

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 fails the exception propagates to {@link EntityChangeLogPoller}, which logs it and
+ *       moves on: the batch is not retried, 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>The batch is dispatched exactly once and is never replayed, so implementations must be
+   * self-healing: recover locally from a failure (for example by clearing the whole cache this
+   * listener maintains, which is a superset of any invalidation it missed) rather than relying on
+   * the poller to retry. If this method throws, the poller logs the failure at {@code ERROR} and
+   * advances its cursor, which leaves the listener's state permanently stale.
    *
    * @param changes the entity changes fetched in one poller cycle
    */