This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new f3a2551a61 [#11736] fix(core): retry failed entity change log
listeners and do not remove logs when polling. (#11739)
f3a2551a61 is described below
commit f3a2551a61c42101e59b013cc354be50d8db4bca
Author: Qi Yu <[email protected]>
AuthorDate: Wed Jul 29 11:06:24 2026 +0800
[#11736] fix(core): retry failed entity change log listeners and do not
remove logs when polling. (#11739)
### What changes were proposed in this pull request?
This PR makes `EntityChangeLogPoller` retry a batch when a listener
fails, with a bounded number of retries:
- Keep the fetched batch until every registered listener applies it.
- Retry only listeners that have not succeeded, and stop waiting for one
that is unregistered.
- Advance the high-water cursor after all pending listeners succeed.
- Bound the retries with `gravitino.entityChangeLog.listenerMaxRetries`
(default 10). Once a listener exhausts them,
`gravitino.entityChangeLog.listenerFailureAction` decides what happens:
`EXIT` (default) stops this server, because its local caches are known
to be stale; `SKIP` drops the batch for that listener and keeps serving.
`EXIT` goes through `System.exit`, so the server shutdown hook still
stops the server gracefully.
- Keep `CatalogChangeLogListener` swallowing its own failures instead of
propagating them. `CatalogManager.consumeLocalMutation` is single-shot,
so re-delivering an already-applied batch would invalidate a catalog
this process mutated itself and close its in-use `IsolatedClassLoader`.
A dropped invalidation only costs staleness bounded by the catalog cache
TTL.
- Increase the fetch batch size from 500 to 2000.
- Log poller startup, consumed ID ranges, cursor advancement, retries,
and listener failures.
- Move cleanup out of the polling path into a dedicated single-threaded
cleaner, and run its first pass after a short randomized delay so that a
frequently restarted server still prunes and HA nodes do not delete the
same rows at once.
- Retain change logs for 30 days by default and run cleanup daily.
- Use database time for both insertion and expiration, and commit each
1000-row cleanup batch.
- Close every entity store component even if an earlier one fails or was
never created.
This PR intentionally does not address the commit-ordering gap or
cache-expiration behavior tracked by #11736.
### Why are the changes needed?
Previously, the poller advanced its cursor even when a listener threw,
permanently dropping that batch for the failed listener. Retrying
without a bound has the opposite problem: a permanently failing listener
would freeze cache invalidation for the whole process, so the retries
are now bounded and the outcome is explicit. Cleanup was coupled to
polling, used a non-committing session (so deletions were never
committed), and calculated expiration from each server's JVM clock.
Fixed: #11736.
### Does this PR introduce _any_ user-facing change?
Yes.
- New config `gravitino.entityChangeLog.listenerMaxRetries` (default
`10`).
- New config `gravitino.entityChangeLog.listenerFailureAction` (default
`EXIT`). With the default, a server whose change log listener keeps
failing after its retries stops itself instead of serving from caches it
knows to be stale. Set it to `SKIP` to prefer availability.
- The default change-log retention increases from 1 day to 30 days, and
the default cleanup interval changes from 1 hour to 1 day.
Both new configs are documented in `docs/gravitino-server-config.md`.
### How was this patch tested?
- `./gradlew :core:test -PskipITs -PskipDockerTests=false`
- Added coverage for failed-listener retries, the retry bound under both
`EXIT` and `SKIP`, a 2001-record backlog, a paused batch blocking new
fetches, catalog-listener failure containment, the cleaner lifecycle and
its initial delay, independent batched cleanup, database-time SQL, and
committed cleanup visibility from a new SQL session.
- The pruning statements are exercised against H2. The PostgreSQL
variant is only asserted at the SQL-string level, matching the existing
PostgreSQL provider tests in `core`; verifying it against a real
PostgreSQL would need a new docker-tagged integration test.
---
.../fileset/TestFilesetCatalogOperations.java | 4 +
.../catalog/kafka/TestKafkaCatalogOperations.java | 4 +
.../generic/TestGenericCatalogOperations.java | 4 +
.../catalog/model/TestModelCatalogOperations.java | 4 +
.../cache/it/AbstractEntityStorageBenchmark.java | 4 +
.../main/java/org/apache/gravitino/Configs.java | 37 ++-
.../catalog/CatalogChangeLogListener.java | 29 +-
.../apache/gravitino/catalog/CatalogManager.java | 4 +-
.../storage/relational/EntityChangeLogCleaner.java | 173 ++++++++++
.../relational/EntityChangeLogListener.java | 4 +
.../storage/relational/EntityChangeLogPoller.java | 354 +++++++++++++++++----
.../storage/relational/RelationalEntityStore.java | 57 +++-
.../relational/mapper/EntityChangeLogMapper.java | 5 +-
.../mapper/EntityChangeLogSQLProviderFactory.java | 4 +-
.../mapper/provider/DatabaseTimeSQL.java | 47 +++
.../base/EntityChangeLogBaseSQLProvider.java | 42 ++-
.../EntityChangeLogPostgreSQLProvider.java | 19 +-
.../authorization/TestAccessControlManager.java | 4 +
.../gravitino/authorization/TestOwnerManager.java | 4 +
.../catalog/TestCatalogChangeLogListener.java | 92 ++++++
.../gravitino/hook/TestFilesetHookDispatcher.java | 4 +
.../apache/gravitino/policy/TestPolicyManager.java | 4 +
.../gravitino/stats/TestStatisticManager.java | 4 +
.../storage/AbstractEntityStorageTest.java | 4 +
.../relational/TestEntityChangeLogCleaner.java | 185 +++++++++++
.../relational/TestEntityChangeLogPoller.java | 257 ++++++++++-----
.../provider/base/TestEntityChangeLogMapper.java | 51 ++-
.../org/apache/gravitino/tag/TestTagManager.java | 4 +
docs/gravitino-server-config.md | 6 +-
29 files changed, 1217 insertions(+), 197 deletions(-)
diff --git
a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java
index 77dbf1f319..1a55d19792 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,6 +20,8 @@ package org.apache.gravitino.catalog.fileset;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -244,6 +246,8 @@ public class TestFilesetCatalogOperations {
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
// Fix cache config for test
diff --git
a/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
b/catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java
index 16b4951236..dd30df96f3 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,6 +21,8 @@ package org.apache.gravitino.catalog.kafka;
import static org.apache.gravitino.Catalog.Type.MESSAGING;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -163,6 +165,8 @@ public class TestKafkaCatalogOperations extends
KafkaClusterEmbedded {
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
// Fix cache config for test
diff --git
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java
index fe749a18a3..c1baa184bf 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,6 +20,8 @@ package org.apache.gravitino.catalog.lakehouse.generic;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -108,6 +110,8 @@ public class TestGenericCatalogOperations {
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(false);
diff --git
a/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java
b/catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java
index 3237e96105..a95892b088 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,6 +20,8 @@ package org.apache.gravtitino.catalog.model;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -127,6 +129,8 @@ public class TestModelCatalogOperations {
when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 *
60L);
// Fix cache config for test
diff --git
a/core/src/jmh/java/org/apache/gravitino/cache/it/AbstractEntityStorageBenchmark.java
b/core/src/jmh/java/org/apache/gravitino/cache/it/AbstractEntityStorageBenchmark.java
index ba733f5f16..3fa058e3b5 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,6 +21,8 @@ package org.apache.gravitino.cache.it;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -294,6 +296,8 @@ public class AbstractEntityStorageBenchmark<E extends
Entity & HasIdentifier> {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index 08702ae734..fb1b0a0c7c 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -20,6 +20,7 @@ package org.apache.gravitino;
import com.google.common.collect.Lists;
import java.io.File;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
@@ -29,6 +30,7 @@ import org.apache.gravitino.config.ConfigBuilder;
import org.apache.gravitino.config.ConfigConstants;
import org.apache.gravitino.config.ConfigEntry;
import org.apache.gravitino.stats.storage.JdbcPartitionStatisticStorageFactory;
+import org.apache.gravitino.storage.relational.EntityChangeLogPoller;
import org.apache.gravitino.utils.FileFetcher;
import org.apache.gravitino.utils.HierarchicalSchemaUtil;
@@ -186,8 +188,10 @@ public class Configs {
.createWithDefault(60 * 60 * 1000L);
public static final long DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS = 3L;
- public static final long DEFAULT_ENTITY_CHANGE_LOG_RETENTION_SECS = 24 * 60
* 60L;
- public static final long DEFAULT_ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS =
60 * 60L;
+ public static final 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;
public static final ConfigEntry<Long> ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS =
new ConfigBuilder("gravitino.entityChangeLog.pollIntervalSecs")
@@ -197,6 +201,31 @@ public class Configs {
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS);
+ public static final ConfigEntry<Integer>
ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES =
+ new ConfigBuilder("gravitino.entityChangeLog.listenerMaxRetries")
+ .doc(
+ "The number of times the poller retries a change log batch for a
failing listener"
+ + " before applying
gravitino.entityChangeLog.listenerFailureAction")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value >= 0,
ConfigConstants.NON_NEGATIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES);
+
+ public static final ConfigEntry<String>
ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION =
+ new ConfigBuilder("gravitino.entityChangeLog.listenerFailureAction")
+ .doc(
+ "What the poller does when a listener exhausted its retries:
EXIT stops this server"
+ + " because its local caches are known to be stale, SKIP
drops the batch for that"
+ + " listener and keeps serving")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .stringConf()
+ .checkValue(
+ value ->
+
Arrays.stream(EntityChangeLogPoller.ListenerFailureAction.values())
+ .anyMatch(action ->
action.name().equalsIgnoreCase(value)),
+ "The value must be either EXIT or SKIP")
+
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION);
+
public static final ConfigEntry<Long> ENTITY_CHANGE_LOG_RETENTION_SECS =
new ConfigBuilder("gravitino.entityChangeLog.retentionSecs")
.doc("The retention time in seconds for entity change logs. Set 0 to
disable cleanup")
@@ -207,7 +236,9 @@ public class Configs {
public static final ConfigEntry<Long>
ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS =
new ConfigBuilder("gravitino.entityChangeLog.cleanupIntervalSecs")
- .doc("The interval in seconds for pruning expired entity change
logs")
+ .doc(
+ "The interval in seconds for independently cleaning expired
entity change logs on a"
+ + " dedicated thread")
.version(ConfigConstants.VERSION_1_3_0)
.longConf()
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
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 218e2d1319..9d7443ca88 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java
@@ -33,6 +33,13 @@ import org.slf4j.LoggerFactory;
*
* <p>This listener is called <em>synchronously</em> in the poller thread.
Implementations must not
* block or perform expensive I/O; only fast, in-memory cache invalidations
are permitted.
+ *
+ * <p>This listener never propagates a failure to the poller, so the poller
never retries a batch
+ * for it. That is deliberate: local-mutation de-duplication ({@link
+ * CatalogManager#consumeLocalMutation}) is single-shot, so re-delivering an
already-applied batch
+ * would invalidate a catalog this process mutated itself and close its
still-in-use {@code
+ * IsolatedClassLoader}. Dropping an invalidation is the cheaper failure: the
catalog cache expires
+ * on access, so staleness is bounded by {@code
gravitino.catalog.cache.evictionIntervalMs}.
*/
public class CatalogChangeLogListener implements EntityChangeLogListener {
@@ -64,17 +71,33 @@ public class CatalogChangeLogListener implements
EntityChangeLogListener {
NameIdentifier ident = identOpt.get();
if (catalogManager.consumeLocalMutation(ident)) {
- LOG.debug("Skipping catalog cache invalidation for local mutation:
{}", ident);
+ LOG.debug(
+ "Skipping catalog cache invalidation for local mutation: {},
change log id {}",
+ ident,
+ change.getId());
continue;
}
- LOG.debug("Invalidating catalog cache due to entity change log: {}",
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());
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: fullName={},
entityType={}",
+ "Failed to process catalog change log record: id={}, fullName={},
entityType={}, "
+ + "operateType={}",
+ change.getId(),
change.getFullName(),
change.getEntityType(),
+ change.getOperateType(),
e);
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index 35d0c3fdb5..17b0d58bcf 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -461,7 +461,9 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
if (!trackLocalMutations) {
return;
}
- localMutationCounts.computeIfAbsent(ident, k -> new
AtomicInteger()).incrementAndGet();
+ int pending =
+ localMutationCounts.computeIfAbsent(ident, k -> new
AtomicInteger()).incrementAndGet();
+ LOG.debug("Marked a local mutation for catalog {}, {} pending marker(s)",
ident, pending);
}
/**
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
new file mode 100644
index 0000000000..29c3ca6bd1
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogCleaner.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Periodically removes expired rows from {@code entity_change_log} on a
dedicated thread.
+ *
+ * <p>Cleanup is intentionally independent from polling. The database
calculates expiration using
+ * its own clock, which is also the clock used when change records are
inserted.
+ *
+ * <p>Retention is validated against the poll interval. A retention shorter
than a few poll cycles
+ * lets this cleaner delete records before every node had a chance to consume
them, which loses
+ * invalidations silently, so such a configuration is rejected at startup
instead.
+ */
+public class EntityChangeLogCleaner implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(EntityChangeLogCleaner.class);
+
+ /** Upper bound for the delay before the first cleanup run. */
+ private static final long MAX_INITIAL_DELAY_MS =
TimeUnit.MINUTES.toMillis(10);
+
+ /**
+ * 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.
+ */
+ private static final long MIN_RETENTION_POLL_CYCLES = 10;
+
+ private final long retentionMs;
+ private final long cleanupIntervalMs;
+
+ private ScheduledExecutorService scheduler;
+
+ /**
+ * Creates an entity change log cleaner.
+ *
+ * @param retentionMs retention time in milliseconds, or 0 to disable cleanup
+ * @param cleanupIntervalMs interval between cleanup runs in milliseconds
+ * @param pollIntervalMs interval between poller cycles in milliseconds,
used to reject a
+ * retention that would drop records before the pollers can consume them
+ */
+ public EntityChangeLogCleaner(long retentionMs, long cleanupIntervalMs, long
pollIntervalMs) {
+ Preconditions.checkArgument(retentionMs >= 0, "retentionMs must be
non-negative");
+ Preconditions.checkArgument(cleanupIntervalMs > 0, "cleanupIntervalMs must
be positive");
+ Preconditions.checkArgument(pollIntervalMs > 0, "pollIntervalMs must be
positive");
+ // Divide rather than multiply: pollIntervalMs comes from configuration
and a multiplication
+ // could overflow.
+ Preconditions.checkArgument(
+ retentionMs == 0 || retentionMs / MIN_RETENTION_POLL_CYCLES >=
pollIntervalMs,
+ "retentionMs (%s) must be 0 to disable cleanup, or at least %s times
the poll interval "
+ + "(%s ms); a shorter retention can prune change records before
every node consumed "
+ + "them and silently lose cache invalidations",
+ retentionMs,
+ MIN_RETENTION_POLL_CYCLES,
+ pollIntervalMs);
+ this.retentionMs = retentionMs;
+ this.cleanupIntervalMs = cleanupIntervalMs;
+ }
+
+ /** Starts the dedicated cleanup thread when automatic cleanup is enabled. */
+ public void start() {
+ if (retentionMs == 0) {
+ LOG.info("Automatic entity change log cleanup is disabled");
+ return;
+ }
+
+ scheduler =
+ Executors.newSingleThreadScheduledExecutor(
+ runnable -> {
+ Thread thread = new Thread(runnable);
+ thread.setName("Gravitino-EntityChangeLogCleaner");
+ thread.setDaemon(true);
+ return thread;
+ });
+ long initialDelayMs = initialDelayMs();
+ scheduler.scheduleWithFixedDelay(
+ this::cleanExpiredChanges, initialDelayMs, cleanupIntervalMs,
TimeUnit.MILLISECONDS);
+ LOG.info(
+ "Starting entity change log cleaner with retention {} ms, cleanup
interval {} ms and "
+ + "initial delay {} ms",
+ retentionMs,
+ cleanupIntervalMs,
+ initialDelayMs);
+ }
+
+ /**
+ * Returns a randomized short delay before the first run. It is deliberately
much smaller than the
+ * cleanup interval, otherwise a server restarted more often than that
interval would never clean
+ * up. The randomization spreads the first run of the HA nodes, which all
delete from the same
+ * rows.
+ */
+ @VisibleForTesting
+ long initialDelayMs() {
+ long bound = Math.min(cleanupIntervalMs, MAX_INITIAL_DELAY_MS);
+ return ThreadLocalRandom.current().nextLong(bound) + 1;
+ }
+
+ @VisibleForTesting
+ void cleanExpiredChanges() {
+ if (retentionMs == 0) {
+ return;
+ }
+
+ long totalPrunedRows = 0;
+ try {
+ int prunedRows;
+ do {
+ prunedRows =
+ SessionUtils.doWithCommitAndFetchResult(
+ EntityChangeLogMapper.class, mapper ->
mapper.pruneOldEntityChanges(retentionMs));
+ totalPrunedRows += prunedRows;
+ } while (prunedRows ==
EntityChangeLogMapper.ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE
+ && !Thread.currentThread().isInterrupted());
+
+ if (totalPrunedRows > 0) {
+ LOG.info(
+ "Pruned {} entity change log record(s) older than {} ms, measured
with database time",
+ totalPrunedRows,
+ retentionMs);
+ } else {
+ // Without this line a cleaner that never runs looks exactly like one
that finds nothing.
+ LOG.debug("No entity change log record older than {} ms to prune",
retentionMs);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to clean expired entity change logs after pruning {}
record(s)",
+ totalPrunedRows,
+ e);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (scheduler != null) {
+ scheduler.shutdown();
+ try {
+ if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+ scheduler.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ scheduler.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+}
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 bd9d54601b..b57d24310b 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,6 +28,10 @@ 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.
+ *
* @param changes the entity changes fetched in one poller cycle
*/
void onEntityChange(List<EntityChangeRecord> changes);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
index 5719cea9f6..34c7e04b1c 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,13 +20,13 @@ package org.apache.gravitino.storage.relational;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.function.LongSupplier;
+import javax.annotation.Nullable;
import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -37,71 +37,101 @@ import org.slf4j.LoggerFactory;
* Global poller for {@code entity_change_log}.
*
* <p>The poller owns the single high-water mark for a Gravitino server
process and dispatches each
- * consumed batch to registered listeners. Listeners should only perform
idempotent local cache
- * invalidation. The cursor always advances after dispatch regardless of
individual listener
- * failures, so a faulty listener cannot block other listeners or prevent
pruning.
+ * consumed batch to registered listeners. The cursor advances only after
every listener applies the
+ * batch. If a listener throws, the immutable batch stays in memory and the
next poll retries only
+ * the listeners that have not succeeded yet.
+ *
+ * <p>A listener that throws after partially applying a batch receives the
whole batch again, so
+ * listeners must make each callback atomic or tolerate retrying changes they
already applied. A
+ * listener that cannot satisfy that contract must swallow its own failures,
as {@code
+ * CatalogChangeLogListener} does.
+ *
+ * <p>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.
*/
public class EntityChangeLogPoller implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(EntityChangeLogPoller.class);
- /** Max entity-change rows to fetch per poller cycle. */
- private static final int ENTITY_CHANGE_POLLER_MAX_ROWS = 500;
+ /**
+ * 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.
+ */
+ private static final int ENTITY_CHANGE_POLLER_MAX_ROWS = 2000;
+
+ /** Max records rendered in a batch summary log line. */
+ private static final int MAX_SUMMARIZED_RECORDS = 20;
+
+ /** What the poller does when a listener keeps failing after {@code
maxListenerRetries}. */
+ public enum ListenerFailureAction {
+ /** Stop this server process, because its local caches are known to be
stale. */
+ EXIT,
+ /** Drop the failing listener from the batch, advance the cursor and keep
serving. */
+ SKIP
+ }
private final List<EntityChangeLogListener> listeners = new
CopyOnWriteArrayList<>();
private final long pollIntervalSecs;
- private final long retentionMs;
- private final long cleanupIntervalMs;
- private final LongSupplier clockMs;
+ private final int maxListenerRetries;
+ private final ListenerFailureAction listenerFailureAction;
+ private final Runnable exitHandler;
+
+ @Nullable private BatchDelivery pendingDelivery;
private ScheduledExecutorService scheduler;
private volatile long entityPollHighWaterId = 0;
- private volatile long lastCleanupMs = Long.MIN_VALUE;
-
- /**
- * Creates an {@link EntityChangeLogPoller}.
- *
- * @param pollIntervalSecs interval between successive polling cycles
- */
- public EntityChangeLogPoller(long pollIntervalSecs) {
- this(
- pollIntervalSecs,
- TimeUnit.DAYS.toMillis(1),
- TimeUnit.HOURS.toMillis(1),
- System::currentTimeMillis);
- }
/**
* Creates an {@link EntityChangeLogPoller}.
*
* @param pollIntervalSecs interval between successive polling cycles
- * @param retentionMs entity change retention in milliseconds, or 0 to
disable cleanup
- * @param cleanupIntervalMs interval between successive cleanup attempts in
milliseconds
+ * @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, long retentionMs, long
cleanupIntervalMs) {
- this(pollIntervalSecs, retentionMs, cleanupIntervalMs,
System::currentTimeMillis);
+ 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, long retentionMs, long cleanupIntervalMs,
LongSupplier clockMs) {
+ long pollIntervalSecs,
+ int maxListenerRetries,
+ ListenerFailureAction listenerFailureAction,
+ Runnable exitHandler) {
Preconditions.checkArgument(pollIntervalSecs > 0, "pollIntervalSecs must
be positive");
- Preconditions.checkArgument(retentionMs >= 0, "retentionMs must be
non-negative");
- Preconditions.checkArgument(cleanupIntervalMs > 0, "cleanupIntervalMs must
be positive");
+ Preconditions.checkArgument(maxListenerRetries >= 0, "maxListenerRetries
must be non-negative");
+ Preconditions.checkArgument(
+ listenerFailureAction != null, "listenerFailureAction cannot be null");
this.pollIntervalSecs = pollIntervalSecs;
- this.retentionMs = retentionMs;
- this.cleanupIntervalMs = cleanupIntervalMs;
- this.clockMs = clockMs;
+ this.maxListenerRetries = maxListenerRetries;
+ this.listenerFailureAction = listenerFailureAction;
+ this.exitHandler = exitHandler;
}
/**
* Registers a listener to receive future entity change batches.
*
+ * <p>A listener only receives batches fetched after it was registered. In
particular, if a batch
+ * is currently paused by a failing listener, the newly registered listener
does not receive that
+ * batch and the cursor moves past it once the batch completes.
+ *
* @param listener the listener to register
*/
public void registerListener(EntityChangeLogListener listener) {
Preconditions.checkArgument(listener != null, "listener cannot be null");
listeners.add(listener);
+ LOG.info(
+ "Registered entity change log listener {}, {} listener(s) active",
+ listener.getClass().getName(),
+ listeners.size());
}
/**
@@ -111,7 +141,12 @@ public class EntityChangeLogPoller implements
AutoCloseable {
*/
public void unregisterListener(EntityChangeLogListener listener) {
Preconditions.checkArgument(listener != null, "listener cannot be null");
- listeners.remove(listener);
+ if (listeners.remove(listener)) {
+ LOG.info(
+ "Unregistered entity change log listener {}, {} listener(s) active",
+ listener.getClass().getName(),
+ listeners.size());
+ }
}
/**
@@ -128,6 +163,14 @@ public class EntityChangeLogPoller implements
AutoCloseable {
getOrDefault(
SessionUtils.getWithoutCommit(
EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
+ LOG.info(
+ "Starting entity change log poller at high-water id {} with a {}
second interval, "
+ + "{} listener(s) registered, maxListenerRetries={},
listenerFailureAction={}",
+ entityPollHighWaterId,
+ pollIntervalSecs,
+ listeners.size(),
+ maxListenerRetries,
+ listenerFailureAction);
scheduler =
Executors.newSingleThreadScheduledExecutor(
@@ -154,6 +197,19 @@ public class EntityChangeLogPoller implements
AutoCloseable {
Thread.currentThread().interrupt();
}
}
+
+ // 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
+ + "]");
}
@VisibleForTesting
@@ -164,35 +220,79 @@ public class EntityChangeLogPoller implements
AutoCloseable {
if (handleInterruptIfAny(e, "Entity change poll")) {
return;
}
- LOG.warn("Entity change poll failed", e);
+ LOG.warn("Entity change poll failed at high-water id {}",
entityPollHighWaterId, e);
}
}
private synchronized void doPollChanges() {
- List<EntityChangeRecord> changes = fetchEntityChanges();
- if (changes.isEmpty()) {
- pruneExpiredChangesIfNeeded();
+ 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;
}
- long maxSeenId = entityPollHighWaterId;
- for (EntityChangeRecord change : changes) {
- if (change.getId() > maxSeenId) {
- maxSeenId = change.getId();
- }
+ delivery = fetchNextDelivery();
+ if (delivery != null) {
+ deliver(delivery);
}
+ }
- List<EntityChangeRecord> dispatchedChanges =
Collections.unmodifiableList(changes);
- for (EntityChangeLogListener listener : listeners) {
- try {
- listener.onEntityChange(dispatchedChanges);
- } catch (Exception e) {
- LOG.warn("Entity change listener {} failed",
listener.getClass().getName(), e);
- }
+ @Nullable
+ private BatchDelivery fetchNextDelivery() {
+ List<EntityChangeRecord> changes = fetchEntityChanges();
+ if (changes.isEmpty()) {
+ return null;
}
- entityPollHighWaterId = maxSeenId;
- pruneExpiredChangesIfNeeded();
+ List<EntityChangeRecord> immutableChanges = List.copyOf(changes);
+ long lastChangeId = immutableChanges.get(immutableChanges.size() -
1).getId();
+ BatchDelivery delivery =
+ new BatchDelivery(immutableChanges, lastChangeId,
List.copyOf(listeners), 1);
+ LOG.debug(
+ "Fetched {} entity change log record(s) after cursor {}, id range [{},
{}]: {}",
+ immutableChanges.size(),
+ entityPollHighWaterId,
+ delivery.firstChangeId(),
+ delivery.lastChangeId,
+ summarize(immutableChanges));
+ return delivery;
+ }
+
+ /**
+ * Renders a batch as {@code id:ENTITY_TYPE:OPERATE_TYPE:fullName} entries,
so a DEBUG log answers
+ * "which invalidations did this node actually see" without querying the
database. Long batches
+ * are truncated, because the whole batch shares one id range that is
already logged.
+ */
+ private static String summarize(List<EntityChangeRecord> changes) {
+ StringBuilder builder = new StringBuilder();
+ int limit = Math.min(changes.size(), MAX_SUMMARIZED_RECORDS);
+ for (int i = 0; i < limit; i++) {
+ EntityChangeRecord change = changes.get(i);
+ if (i > 0) {
+ builder.append(", ");
+ }
+ builder
+ .append(change.getId())
+ .append(':')
+ .append(change.getEntityType())
+ .append(':')
+ .append(change.getOperateType())
+ .append(':')
+ .append(change.getFullName());
+ }
+ if (changes.size() > limit) {
+ builder.append(", ... ").append(changes.size() - limit).append(" more");
+ }
+ return builder.toString();
}
private List<EntityChangeRecord> fetchEntityChanges() {
@@ -218,31 +318,145 @@ public class EntityChangeLogPoller implements
AutoCloseable {
return false;
}
- private void pruneExpiredChangesIfNeeded() {
- if (retentionMs <= 0) {
+ private static long getOrDefault(Long value) {
+ return value == null ? 0L : value;
+ }
+
+ private void deliver(BatchDelivery delivery) {
+ List<EntityChangeLogListener> failedListeners = notifyListeners(delivery);
+ if (failedListeners.isEmpty()) {
+ advanceCursor(delivery);
return;
}
- long now = clockMs.getAsLong();
- if (lastCleanupMs != Long.MIN_VALUE && now - lastCleanupMs <
cleanupIntervalMs) {
+ if (delivery.attempts > maxListenerRetries) {
+ handleExhaustedRetries(delivery, failedListeners);
return;
}
- long before = now - retentionMs;
- try {
- SessionUtils.doWithoutCommit(
- EntityChangeLogMapper.class, mapper ->
mapper.pruneOldEntityChanges(before));
- } catch (Exception e) {
- LOG.warn("Failed to prune expired entity change logs before {}", before,
e);
- } finally {
- // Always advance the cursor regardless of success or failure. A
transient DB error
- // should not cause repeated prune attempts on every poll cycle (every
few seconds)
- // until one eventually succeeds — the next cleanup will happen after
cleanupIntervalMs.
- lastCleanupMs = now;
+ 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);
+ }
+
+ 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",
+ delivery.changes.size(),
+ delivery.firstChangeId(),
+ delivery.lastChangeId,
+ previousHighWaterId,
+ entityPollHighWaterId,
+ delivery.approximateLagMs());
+ }
+
+ private void handleExhaustedRetries(
+ BatchDelivery delivery, List<EntityChangeLogListener> failedListeners) {
+ List<String> failedListenerNames = new ArrayList<>();
+ for (EntityChangeLogListener listener : failedListeners) {
+ failedListenerNames.add(listener.getClass().getName());
+ }
+
+ if (listenerFailureAction == ListenerFailureAction.EXIT) {
+ LOG.error(
+ "Stopping this server: listener(s) {} failed to apply entity change
log batch id range "
+ + "[{}, {}] after {} attempt(s), so local caches are stale and
cannot be trusted",
+ failedListenerNames,
+ delivery.firstChangeId(),
+ delivery.lastChangeId,
+ delivery.attempts);
+ exitHandler.run();
+ return;
}
+
+ LOG.error(
+ "Dropping entity change log batch id range [{}, {}] for listener(s) {}
after {} attempt(s);"
+ + " their local caches may be stale until the affected entries
expire",
+ delivery.firstChangeId(),
+ delivery.lastChangeId,
+ failedListenerNames,
+ delivery.attempts);
+ advanceCursor(delivery);
}
- private static long getOrDefault(Long value) {
- return value == null ? 0L : value;
+ private List<EntityChangeLogListener> notifyListeners(BatchDelivery
delivery) {
+ List<EntityChangeLogListener> failedListeners = new ArrayList<>();
+ for (EntityChangeLogListener listener : delivery.pendingListeners) {
+ if (!listeners.contains(listener)) {
+ LOG.debug(
+ "Skipping unregistered entity change log listener {} for batch id
range [{}, {}]",
+ listener.getClass().getName(),
+ delivery.firstChangeId(),
+ delivery.lastChangeId);
+ continue;
+ }
+
+ try {
+ listener.onEntityChange(delivery.changes);
+ LOG.debug(
+ "Entity change log listener {} consumed batch id range [{}, {}]",
+ 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
[{}, {}]",
+ listener.getClass().getName(),
+ delivery.firstChangeId(),
+ delivery.lastChangeId,
+ e);
+ }
+ }
+ return failedListeners;
+ }
+
+ private static class BatchDelivery {
+ private final List<EntityChangeRecord> changes;
+ private final long lastChangeId;
+ private final List<EntityChangeLogListener> pendingListeners;
+
+ /** How many times this batch has been dispatched, starting at 1 for the
initial dispatch. */
+ private final int attempts;
+
+ private BatchDelivery(
+ List<EntityChangeRecord> changes,
+ long lastChangeId,
+ List<EntityChangeLogListener> pendingListeners,
+ int attempts) {
+ this.changes = changes;
+ this.lastChangeId = lastChangeId;
+ this.pendingListeners = pendingListeners;
+ this.attempts = attempts;
+ }
+
+ private BatchDelivery retryOnly(List<EntityChangeLogListener>
failedListeners) {
+ return new BatchDelivery(changes, lastChangeId,
List.copyOf(failedListeners), attempts + 1);
+ }
+
+ private long firstChangeId() {
+ return changes.get(0).getId();
+ }
+
+ /**
+ * How far this node is behind the newest record of the batch, in
milliseconds. It compares the
+ * DB-generated {@code created_at} with the local JVM clock, so it is only
an estimate and can
+ * even be negative when the two clocks disagree. It is still the quickest
way to spot a node
+ * that fell behind: a steadily growing value means this node is not
keeping up.
+ */
+ private long approximateLagMs() {
+ return System.currentTimeMillis() - changes.get(changes.size() -
1).getCreatedAt();
+ }
}
}
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 a45d4b7d96..f14f41c9a1 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
@@ -26,10 +26,12 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.Config;
import org.apache.gravitino.Configs;
@@ -77,6 +79,7 @@ public class RelationalEntityStore
private RelationalBackend backend;
private RelationalGarbageCollector garbageCollector;
private EntityChangeLogPoller entityChangeLogPoller;
+ private EntityChangeLogCleaner entityChangeLogCleaner;
private EntityCache cache;
@VisibleForTesting
@@ -99,16 +102,23 @@ public class RelationalEntityStore
this.garbageCollector = new RelationalGarbageCollector(backend, config);
this.garbageCollector.start();
- // The change-log poller is a side module of the entity store: it polls
the entity_change_log
- // table this store writes to, dispatches batches to registered listeners
(e.g. for cross-node
- // cache invalidation), and prunes expired rows. Like the garbage
collector, it is owned and
- // lifecycle-managed by the store itself.
+ // 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)));
+ this.entityChangeLogCleaner =
+ new EntityChangeLogCleaner(
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)),
-
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)));
+
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)),
+
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)));
this.entityChangeLogPoller.start();
+ this.entityChangeLogCleaner.start();
}
private RelationalBackend createRelationalEntityBackend(Config config) {
@@ -276,10 +286,39 @@ public class RelationalEntityStore
@Override
public void close() throws IOException {
- cache.clear();
- entityChangeLogPoller.close();
- garbageCollector.close();
- backend.close();
+ // Keep shutting the remaining components down even if one of them fails,
and tolerate a
+ // half-finished initialize() that left some of them null.
+ IOException failure = null;
+ failure = closeComponent(failure, "entity cache", cache == null ? null :
cache::clear);
+ failure = closeComponent(failure, "entity change log poller",
entityChangeLogPoller);
+ failure = closeComponent(failure, "entity change log cleaner",
entityChangeLogCleaner);
+ failure = closeComponent(failure, "relational garbage collector",
garbageCollector);
+ failure = closeComponent(failure, "relational backend", backend);
+
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private static IOException closeComponent(
+ @Nullable IOException failure, String name, @Nullable AutoCloseable
component) {
+ if (component == null) {
+ return failure;
+ }
+
+ try {
+ component.close();
+ return failure;
+ } catch (Exception e) {
+ LOGGER.warn("Failed to close {}", name, e);
+ if (failure != null) {
+ failure.addSuppressed(e);
+ return failure;
+ }
+ return e instanceof IOException
+ ? (IOException) e
+ : new IOException("Failed to close " + name, e);
+ }
}
@Override
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
index d82754947c..b82faceb7e 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
@@ -36,6 +36,9 @@ public interface EntityChangeLogMapper {
String ENTITY_CHANGE_LOG_TABLE_NAME = "entity_change_log";
+ /** Max expired rows removed in one cleanup transaction. */
+ int ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE = 1000;
+
@SelectProvider(type = EntityChangeLogSQLProviderFactory.class, method =
"selectEntityChanges")
List<EntityChangeRecord> selectEntityChanges(
@Param("lastConsumedId") long lastConsumedId, @Param("maxRows") int
maxRows);
@@ -51,5 +54,5 @@ public interface EntityChangeLogMapper {
@Param("operateType") OperateType operateType);
@DeleteProvider(type = EntityChangeLogSQLProviderFactory.class, method =
"pruneOldEntityChanges")
- void pruneOldEntityChanges(@Param("before") long before);
+ int pruneOldEntityChanges(@Param("retentionMs") long retentionMs);
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
index c48f2c0a1a..36620bd905 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
@@ -67,7 +67,7 @@ public class EntityChangeLogSQLProviderFactory {
return getProvider().insertEntityChange(metalakeName, entityType,
fullName, operateType);
}
- public static String pruneOldEntityChanges(@Param("before") long before) {
- return getProvider().pruneOldEntityChanges(before);
+ public static String pruneOldEntityChanges(@Param("retentionMs") long
retentionMs) {
+ return getProvider().pruneOldEntityChanges(retentionMs);
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DatabaseTimeSQL.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DatabaseTimeSQL.java
new file mode 100644
index 0000000000..65d8f6b547
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DatabaseTimeSQL.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper.provider;
+
+/**
+ * SQL expressions that evaluate to the current time in epoch milliseconds,
using the database
+ * clock.
+ *
+ * <p>Prefer these over a timestamp computed in the JVM whenever a value is
written to or compared
+ * against a millisecond column: in a multi-node deployment every server has
its own clock, while
+ * the database provides a single clock all of them agree on.
+ *
+ * <p>The expressions live here, rather than in one SQL provider, so that
every provider can reuse
+ * them. Most providers still inline the MySQL flavour; they can be migrated
to {@link #MYSQL}
+ * separately, since changing them is unrelated to any single feature.
+ */
+public final class DatabaseTimeSQL {
+
+ /**
+ * MySQL flavour, also used by H2 in {@code MODE=MYSQL}. This is the
expression that 30+ SQL
+ * providers currently inline.
+ */
+ public static final String MYSQL =
+ "((UNIX_TIMESTAMP() * 1000.0) + EXTRACT(MICROSECOND FROM
CURRENT_TIMESTAMP(3)) / 1000)";
+
+ /** PostgreSQL flavour. */
+ public static final String POSTGRESQL =
+ "CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)";
+
+ private DatabaseTimeSQL() {}
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
index 78695830cc..d5c4685efb 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
@@ -18,13 +18,26 @@
*/
package org.apache.gravitino.storage.relational.mapper.provider.base;
+import static
org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper.ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE;
import static
org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper.ENTITY_CHANGE_LOG_TABLE_NAME;
+import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.ibatis.annotations.Param;
public class EntityChangeLogBaseSQLProvider {
+ /**
+ * DB-side expression for "now" in milliseconds; PostgreSQL overrides both
statements that use it
+ * in its own provider.
+ *
+ * <p>Insertion and expiration both use this expression, so retention is
measured entirely with
+ * the database clock and is immune to clock skew between Gravitino nodes.
Round-trip behaviour is
+ * verified by {@code
TestEntityChangeLogMapper#testEntityChangeLogInsertAndSelect}, which asserts
+ * the persisted value is within 1 s of the JVM clock.
+ */
+ private static final String CURRENT_TIME_MILLIS_SQL = DatabaseTimeSQL.MYSQL;
+
/**
* Cursor-advance contract for the entity change poller: {@code id} is
monotonic and unique, so
* callers only need to remember the last consumed id.
@@ -49,15 +62,7 @@ public class EntityChangeLogBaseSQLProvider {
return "SELECT COALESCE(MAX(id), 0) FROM " + ENTITY_CHANGE_LOG_TABLE_NAME;
}
- /**
- * The {@code (UNIX_TIMESTAMP() * 1000.0) + EXTRACT(MICROSECOND FROM
CURRENT_TIMESTAMP(3)) / 1000}
- * expression is the established codebase convention for DB-generated
millisecond timestamps,
- * shared with 27+ other base providers (TableMetaBaseSQLProvider,
FilesetVersionBaseSQLProvider,
- * etc.). It works on MySQL natively and on H2 in {@code MODE=MYSQL};
PostgreSQL overrides this
- * method in its own provider. Round-trip behaviour is verified by {@code
- * TestEntityChangeLogMapper#testEntityChangeLogInsertAndSelect}, which
asserts the persisted
- * value is within 1 s of the JVM clock.
- */
+ /** Inserts a change record, stamping {@code created_at} with {@link
#CURRENT_TIME_MILLIS_SQL}. */
public String insertEntityChange(
@Param("metalakeName") String metalakeName,
@Param("entityType") String entityType,
@@ -67,16 +72,23 @@ public class EntityChangeLogBaseSQLProvider {
+ ENTITY_CHANGE_LOG_TABLE_NAME
+ " (metalake_name, entity_type, entity_full_name, operate_type,
created_at)"
+ " VALUES (#{metalakeName}, #{entityType}, #{fullName},
#{operateType},"
- + " (UNIX_TIMESTAMP() * 1000.0)"
- + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000)";
+ + CURRENT_TIME_MILLIS_SQL
+ + ")";
}
- public String pruneOldEntityChanges(@Param("before") long before) {
+ public String pruneOldEntityChanges(@Param("retentionMs") long retentionMs) {
// Keep the retention window conservative. A running server can be delayed
by long GC pauses,
- // network isolation, scheduler stalls, or clock skew between nodes;
pruning too aggressively
- // can let that server miss an invalidation while its local cache is still
warm.
+ // network isolation, or scheduler stalls; pruning too aggressively can
let that server miss an
+ // invalidation while its local cache is still warm.
+ //
+ // No ORDER BY here, unlike the PostgreSQL provider: H2's DELETE grammar
accepts LIMIT but not
+ // ORDER BY. Every matched row is expired anyway, so the deletion order
does not matter; the
+ // cleaner randomizes its start time to keep HA nodes from deleting the
same rows at once.
return "DELETE FROM "
+ ENTITY_CHANGE_LOG_TABLE_NAME
- + " WHERE created_at < #{before} LIMIT 1000";
+ + " WHERE created_at < "
+ + CURRENT_TIME_MILLIS_SQL
+ + " - #{retentionMs} LIMIT "
+ + ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE;
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/EntityChangeLogPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/EntityChangeLogPostgreSQLProvider.java
index d6cdbc5851..4f41fe018f 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/EntityChangeLogPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/EntityChangeLogPostgreSQLProvider.java
@@ -18,14 +18,22 @@
*/
package org.apache.gravitino.storage.relational.mapper.provider.postgresql;
+import static
org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper.ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE;
import static
org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper.ENTITY_CHANGE_LOG_TABLE_NAME;
+import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import
org.apache.gravitino.storage.relational.mapper.provider.base.EntityChangeLogBaseSQLProvider;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.ibatis.annotations.Param;
public class EntityChangeLogPostgreSQLProvider extends
EntityChangeLogBaseSQLProvider {
+ /**
+ * PostgreSQL flavour of the DB-side "now in milliseconds" expression.
Insertion and expiration
+ * share it, so retention is measured entirely with the database clock.
+ */
+ private static final String CURRENT_TIME_MILLIS_SQL =
DatabaseTimeSQL.POSTGRESQL;
+
@Override
public String insertEntityChange(
@Param("metalakeName") String metalakeName,
@@ -36,15 +44,20 @@ public class EntityChangeLogPostgreSQLProvider extends
EntityChangeLogBaseSQLPro
+ ENTITY_CHANGE_LOG_TABLE_NAME
+ " (metalake_name, entity_type, entity_full_name, operate_type,
created_at)"
+ " VALUES (#{metalakeName}, #{entityType}, #{fullName},
#{operateType},"
- + " CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT))";
+ + CURRENT_TIME_MILLIS_SQL
+ + ")";
}
@Override
- public String pruneOldEntityChanges(@Param("before") long before) {
+ public String pruneOldEntityChanges(@Param("retentionMs") long retentionMs) {
return "DELETE FROM "
+ ENTITY_CHANGE_LOG_TABLE_NAME
+ " WHERE id IN (SELECT id FROM "
+ ENTITY_CHANGE_LOG_TABLE_NAME
- + " WHERE created_at < #{before} ORDER BY created_at LIMIT 1000)";
+ + " WHERE created_at < "
+ + CURRENT_TIME_MILLIS_SQL
+ + " - #{retentionMs} ORDER BY created_at LIMIT "
+ + ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE
+ + ")";
}
}
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 10813b5e64..8cc94e01fb 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.authorization;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -135,6 +137,8 @@ public class TestAccessControlManager {
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
index 936d5336a7..564000a7ca 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.authorization;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -114,6 +116,8 @@ public class TestOwnerManager {
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java
new file mode 100644
index 0000000000..410dcd5829
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.catalog;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import java.util.List;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.catalog.CatalogManager.CatalogWrapper;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.apache.gravitino.storage.relational.po.cache.OperateType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestCatalogChangeLogListener {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testProcessesRemainingChangesAndSwallowsFailure() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
+ NameIdentifier failedIdentifier = NameIdentifier.of("metalake", "failed");
+ NameIdentifier successfulIdentifier = NameIdentifier.of("metalake",
"successful");
+
+ when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
+ when(catalogManager.consumeLocalMutation(failedIdentifier))
+ .thenThrow(new RuntimeException("invalidation failed"));
+
+ 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"))));
+
+ verify(catalogCache).invalidate(successfulIdentifier);
+ verify(catalogCache, never()).invalidate(failedIdentifier);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testSkipsLocalMutationAndNonCatalogRecords() {
+ CatalogManager catalogManager = mock(CatalogManager.class);
+ Cache<NameIdentifier, CatalogWrapper> catalogCache = mock(Cache.class);
+ NameIdentifier localIdentifier = NameIdentifier.of("metalake", "local");
+
+ when(catalogManager.getCatalogCache()).thenReturn(catalogCache);
+
when(catalogManager.consumeLocalMutation(localIdentifier)).thenReturn(true);
+
+ CatalogChangeLogListener listener = new
CatalogChangeLogListener(catalogManager);
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ listener.onEntityChange(
+ List.of(
+ change(1L, "metalake.local"),
+ change(2L, "metalake"),
+ change(3L, "metalake.cat.schema"),
+ new EntityChangeRecord(
+ 4L, "metalake", "SCHEMA", "metalake.cat.sch",
OperateType.ALTER, 0L))));
+
+ verify(catalogCache, never()).invalidate(any());
+ }
+
+ 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 0a97ec1efb..943444019e 100644
---
a/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.hook;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -218,6 +220,8 @@ public class TestFilesetHookDispatcher extends
TestOperationDispatcher {
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60
* 1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 *
60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git
a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
index 5b80249042..46504db01c 100644
--- a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
+++ b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.policy;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -199,6 +201,8 @@ public class TestPolicyManager {
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git
a/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
b/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
index aa0cb0fea5..a8a7538ba9 100644
--- a/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
+++ b/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.stats;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -113,6 +115,8 @@ public class TestStatisticManager {
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
b/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
index cac53eed61..31e39e3388 100644
---
a/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
+++
b/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
@@ -21,6 +21,8 @@ package org.apache.gravitino.storage;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -147,6 +149,8 @@ abstract class AbstractEntityStorageTest {
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogCleaner.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogCleaner.java
new file mode 100644
index 0000000000..1b13088c51
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogCleaner.java
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+public class TestEntityChangeLogCleaner {
+
+ private static final long RETENTION_MS = TimeUnit.DAYS.toMillis(30);
+ private static final long POLL_INTERVAL_MS = TimeUnit.SECONDS.toMillis(3);
+
+ @Test
+ void testRejectsInvalidConfiguration() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> new EntityChangeLogCleaner(-1, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> new EntityChangeLogCleaner(RETENTION_MS, 0, POLL_INTERVAL_MS));
+ }
+
+ @Test
+ void testRejectsRetentionShorterThanTheConsumptionWindow() {
+ // A retention this short lets the cleaner delete records before every
node polled them, which
+ // would lose invalidations silently, so it must fail at startup instead.
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new EntityChangeLogCleaner(
+ POLL_INTERVAL_MS * 9, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS));
+ Assertions.assertTrue(
+ e.getMessage().contains("at least"), "unexpected message: " +
e.getMessage());
+
+ // Exactly at the bound is accepted, and so is a disabled cleanup with any
poll interval.
+ Assertions.assertDoesNotThrow(
+ () ->
+ new EntityChangeLogCleaner(
+ POLL_INTERVAL_MS * 10, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS));
+ Assertions.assertDoesNotThrow(
+ () -> new EntityChangeLogCleaner(0, TimeUnit.DAYS.toMillis(1),
TimeUnit.DAYS.toMillis(7)));
+ }
+
+ @Test
+ void testDisablesCleanupWhenRetentionIsZero() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ EntityChangeLogCleaner cleaner =
+ new EntityChangeLogCleaner(0, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS);
+
+ cleaner.cleanExpiredChanges();
+ }
+
+ verify(mapper, never()).pruneOldEntityChanges(anyLong());
+ }
+
+ @Test
+ void testFirstRunDoesNotWaitForAWholeCleanupInterval() {
+ // A server restarted more often than the cleanup interval must still
prune, so the first run
+ // is scheduled after a short randomized delay instead of a full interval.
+ EntityChangeLogCleaner dailyCleaner =
+ new EntityChangeLogCleaner(RETENTION_MS, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS);
+ for (int i = 0; i < 100; i++) {
+ long delayMs = dailyCleaner.initialDelayMs();
+ Assertions.assertTrue(delayMs > 0, "initial delay must be positive, got
" + delayMs);
+ Assertions.assertTrue(
+ delayMs <= TimeUnit.MINUTES.toMillis(10), "initial delay too long: "
+ delayMs);
+ }
+
+ // Never longer than the configured interval either.
+ EntityChangeLogCleaner frequentCleaner =
+ new EntityChangeLogCleaner(RETENTION_MS, 50L, POLL_INTERVAL_MS);
+ for (int i = 0; i < 100; i++) {
+ Assertions.assertTrue(frequentCleaner.initialDelayMs() <= 50L);
+ }
+ }
+
+ @Test
+ void testStartAndCloseAreSafeToPair() {
+ EntityChangeLogCleaner cleaner =
+ new EntityChangeLogCleaner(RETENTION_MS, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS);
+
+ Assertions.assertDoesNotThrow(cleaner::start);
+ Assertions.assertDoesNotThrow(cleaner::close);
+ // close() on a cleaner that was never started must not fail either.
+ Assertions.assertDoesNotThrow(
+ () ->
+ new EntityChangeLogCleaner(RETENTION_MS,
TimeUnit.DAYS.toMillis(1), POLL_INTERVAL_MS)
+ .close());
+ }
+
+ @Test
+ void testStartIsNoOpWhenRetentionIsZero() throws Exception {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ try (EntityChangeLogCleaner cleaner = new EntityChangeLogCleaner(0, 1L,
POLL_INTERVAL_MS)) {
+ cleaner.start();
+ Thread.sleep(200);
+ }
+ }
+
+ verify(mapper, never()).pruneOldEntityChanges(anyLong());
+ }
+
+ @Test
+ void testDrainsExpiredRowsInCommittedBatches() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.pruneOldEntityChanges(RETENTION_MS))
+ .thenReturn(EntityChangeLogMapper.ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE)
+ .thenReturn(EntityChangeLogMapper.ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE)
+ .thenReturn(7);
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ EntityChangeLogCleaner cleaner =
+ new EntityChangeLogCleaner(RETENTION_MS, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS);
+
+ cleaner.cleanExpiredChanges();
+ }
+
+ verify(mapper, times(3)).pruneOldEntityChanges(RETENTION_MS);
+ }
+
+ @Test
+ void testCleanupFailureDoesNotEscapeScheduledTask() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.pruneOldEntityChanges(RETENTION_MS))
+ .thenThrow(new RuntimeException("database unavailable"));
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ EntityChangeLogCleaner cleaner =
+ new EntityChangeLogCleaner(RETENTION_MS, TimeUnit.DAYS.toMillis(1),
POLL_INTERVAL_MS);
+
+ Assertions.assertDoesNotThrow(cleaner::cleanExpiredChanges);
+ }
+ }
+
+ private static void mockSessionUtils(
+ MockedStatic<SessionUtils> sessionUtils, EntityChangeLogMapper mapper) {
+ sessionUtils
+ .when(() -> SessionUtils.doWithCommitAndFetchResult(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<Object, Object> function = invocation.getArgument(1);
+ return function.apply(mapper);
+ });
+ }
+}
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 4eea24e24f..bf54d5b6ef 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,18 +19,21 @@
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.TimeUnit;
-import java.util.function.Consumer;
+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;
@@ -41,10 +44,15 @@ import org.mockito.MockedStatic;
public class TestEntityChangeLogPoller {
+ private static final int MAX_ROWS = 2000;
+
@Test
- void testRejectsNonPositivePollInterval() {
- Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(0));
- Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(-1));
+ 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));
}
@Test
@@ -52,22 +60,16 @@ public class TestEntityChangeLogPoller {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
EntityChangeRecord second = change(2L, "SCHEMA", "ml1.cat1.sch1");
- when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(first,
second));
- when(mapper.selectEntityChanges(2L, 500)).thenReturn(List.of());
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(first,
second));
+ when(mapper.selectEntityChanges(2L, MAX_ROWS)).thenReturn(List.of());
List<EntityChangeRecord> firstListenerRecords = new ArrayList<>();
List<EntityChangeRecord> secondListenerRecords = new ArrayList<>();
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
- sessionUtils
- .when(() -> SessionUtils.getWithoutCommit(any(), any()))
- .thenAnswer(
- invocation -> {
- Function<Object, Object> func = invocation.getArgument(1);
- return func.apply(mapper);
- });
-
- EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller = newPoller(1, 10);
poller.registerListener(firstListenerRecords::addAll);
poller.registerListener(secondListenerRecords::addAll);
@@ -77,150 +79,247 @@ public class TestEntityChangeLogPoller {
Assertions.assertEquals(List.of(first, second), firstListenerRecords);
Assertions.assertEquals(List.of(first, second), secondListenerRecords);
+ verify(mapper).selectEntityChanges(2L, MAX_ROWS);
}
@Test
- void testListenerFailureDoesNotBlockOtherListenersAndCursorStillAdvances() {
+ void testRetriesOnlyFailedListenersBeforeAdvancingCursor() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
- when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(change));
- when(mapper.selectEntityChanges(1L, 500)).thenReturn(List.of());
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
+ when(mapper.selectEntityChanges(1L, MAX_ROWS)).thenReturn(List.of());
List<EntityChangeRecord> received = new ArrayList<>();
+ AtomicInteger failingListenerCalls = new AtomicInteger();
+ AtomicBoolean firstCall = new AtomicBoolean(true);
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
- sessionUtils
- .when(() -> SessionUtils.getWithoutCommit(any(), any()))
- .thenAnswer(
- invocation -> {
- Function<Object, Object> func = invocation.getArgument(1);
- return func.apply(mapper);
- });
-
- EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller = newPoller(1, 10);
poller.registerListener(
changes -> {
- throw new RuntimeException("listener failed");
+ failingListenerCalls.incrementAndGet();
+ if (firstCall.getAndSet(false)) {
+ throw new RuntimeException("listener failed");
+ }
});
poller.registerListener(received::addAll);
poller.pollChanges();
poller.pollChanges();
+ 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 testPollChangesCatchesFetchFailures() {
+ void testUnregisteredFailedListenerDoesNotBlockCursor() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- when(mapper.selectEntityChanges(0L, 500)).thenThrow(new
RuntimeException("db failed"));
+ EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
+ when(mapper.selectEntityChanges(1L, MAX_ROWS)).thenReturn(List.of());
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
- sessionUtils
- .when(() -> SessionUtils.getWithoutCommit(any(), any()))
- .thenAnswer(
- invocation -> {
- Function<Object, Object> func = invocation.getArgument(1);
- return func.apply(mapper);
- });
+ mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ EntityChangeLogPoller poller = newPoller(1, 10);
+ EntityChangeLogListener failedListener =
+ changes -> {
+ throw new RuntimeException("listener failed");
+ };
+ poller.registerListener(failedListener);
- Assertions.assertDoesNotThrow(poller::pollChanges);
+ poller.pollChanges();
+ poller.unregisterListener(failedListener);
+ poller.pollChanges();
+ poller.pollChanges();
}
+
+ verify(mapper).selectEntityChanges(1L, MAX_ROWS);
}
@Test
- void testDispatchesImmutableBatchToListeners() {
+ void testPausedBatchBlocksFetchingNewBatches() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
- EntityChangeRecord second = change(2L, "SCHEMA", "ml1.cat1.sch1");
- when(mapper.selectEntityChanges(0L, 500)).thenReturn(new
ArrayList<>(List.of(first, second)));
-
- List<EntityChangeRecord> received = new ArrayList<>();
+ EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
- sessionUtils
- .when(() -> SessionUtils.getWithoutCommit(any(), any()))
- .thenAnswer(
- invocation -> {
- Function<Object, Object> func = invocation.getArgument(1);
- return func.apply(mapper);
- });
- sessionUtils
- .when(() -> SessionUtils.doWithoutCommit(any(), any()))
- .thenAnswer(invocation -> null);
-
- EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller = newPoller(1, 10);
poller.registerListener(
- changes ->
Assertions.assertThrows(UnsupportedOperationException.class, changes::clear));
- poller.registerListener(received::addAll);
+ changes -> {
+ throw new RuntimeException("listener failed");
+ });
poller.pollChanges();
+ poller.pollChanges();
}
- Assertions.assertEquals(List.of(first, second), received);
+ // 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 testPrunesExpiredChangesAfterCleanupInterval() {
+ void testStopsServerAfterListenerExhaustsRetries() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of());
+ EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1");
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(List.of(change));
+
+ AtomicInteger listenerCalls = new AtomicInteger();
+ AtomicInteger exitCalls = new AtomicInteger();
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
EntityChangeLogPoller poller =
- new EntityChangeLogPoller(
- 1, TimeUnit.DAYS.toMillis(1), TimeUnit.HOURS.toMillis(1), () ->
100_000_000L);
+ new EntityChangeLogPoller(1, 1, ListenerFailureAction.EXIT,
exitCalls::incrementAndGet);
+ poller.registerListener(
+ changes -> {
+ listenerCalls.incrementAndGet();
+ throw new RuntimeException("listener failed");
+ });
poller.pollChanges();
+ Assertions.assertEquals(0, exitCalls.get());
+ poller.pollChanges();
}
- verify(mapper).pruneOldEntityChanges(100_000_000L -
TimeUnit.DAYS.toMillis(1));
+ // 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);
}
@Test
- void testSkipsPruneBeforeCleanupInterval() {
+ void testSkipActionDropsFailedListenerAndAdvancesCursor() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- when(mapper.selectEntityChanges(0L, 500)).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 listenerCalls = new AtomicInteger();
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
EntityChangeLogPoller poller =
new EntityChangeLogPoller(
- 1, TimeUnit.DAYS.toMillis(1), TimeUnit.HOURS.toMillis(1), () ->
100_000_000L);
+ 1, 0, ListenerFailureAction.SKIP,
TestEntityChangeLogPoller::failOnExit);
+ poller.registerListener(
+ changes -> {
+ listenerCalls.incrementAndGet();
+ throw new RuntimeException("listener failed");
+ });
poller.pollChanges();
poller.pollChanges();
+ poller.pollChanges();
}
- verify(mapper).pruneOldEntityChanges(100_000_000L -
TimeUnit.DAYS.toMillis(1));
+ // 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());
+ verify(mapper).selectEntityChanges(1L, MAX_ROWS);
+ verify(mapper).selectEntityChanges(2L, MAX_ROWS);
}
@Test
- void testDisablesPruneWhenRetentionIsZero() {
+ void testConsumesBacklogLargerThanOneBatch() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
- when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of());
+ List<EntityChangeRecord> firstBatch = changes(1L, MAX_ROWS);
+ EntityChangeRecord remainingChange = change(2001L, "TABLE",
"ml1.cat1.schema1.table2001");
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenReturn(firstBatch);
+ when(mapper.selectEntityChanges(2000L,
MAX_ROWS)).thenReturn(List.of(remainingChange));
+ List<EntityChangeRecord> received = new ArrayList<>();
try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
mockSessionUtils(sessionUtils, mapper);
- EntityChangeLogPoller poller =
- new EntityChangeLogPoller(1, 0L, TimeUnit.HOURS.toMillis(1), () ->
100_000_000L);
+ EntityChangeLogPoller poller = newPoller(1, 10);
+ poller.registerListener(received::addAll);
poller.pollChanges();
+ poller.pollChanges();
}
- verify(mapper, never()).pruneOldEntityChanges(anyLong());
+ Assertions.assertEquals(2001, received.size());
+ Assertions.assertEquals(remainingChange, received.get(2000));
+ }
+
+ @Test
+ void testPollChangesCatchesFetchFailures() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, MAX_ROWS)).thenThrow(new
RuntimeException("db failed"));
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller = newPoller(1, 10);
+
+ Assertions.assertDoesNotThrow(poller::pollChanges);
+ }
+ }
+
+ @Test
+ void testDispatchesImmutableBatchToListeners() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ EntityChangeRecord first = change(1L, "CATALOG", "ml1.cat1");
+ EntityChangeRecord second = change(2L, "SCHEMA", "ml1.cat1.sch1");
+ when(mapper.selectEntityChanges(0L, MAX_ROWS))
+ .thenReturn(new ArrayList<>(List.of(first, second)));
+
+ List<EntityChangeRecord> received = new ArrayList<>();
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller = newPoller(1, 10);
+ poller.registerListener(
+ changes ->
Assertions.assertThrows(UnsupportedOperationException.class, changes::clear));
+ poller.registerListener(received::addAll);
+
+ poller.pollChanges();
+ }
+
+ 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);
}
+ private static List<EntityChangeRecord> changes(long firstId, long lastId) {
+ List<EntityChangeRecord> changes = new ArrayList<>();
+ for (long id = firstId; id <= lastId; id++) {
+ changes.add(change(id, "TABLE", "ml1.cat1.schema1.table" + id));
+ }
+ return changes;
+ }
+
private static void mockSessionUtils(
MockedStatic<SessionUtils> sessionUtils, EntityChangeLogMapper mapper) {
sessionUtils
@@ -230,13 +329,5 @@ public class TestEntityChangeLogPoller {
Function<Object, Object> func = invocation.getArgument(1);
return func.apply(mapper);
});
- sessionUtils
- .when(() -> SessionUtils.doWithoutCommit(any(), any()))
- .thenAnswer(
- invocation -> {
- Consumer<Object> consumer = invocation.getArgument(1);
- consumer.accept(mapper);
- return null;
- });
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
index 8893a74116..89fa8d6543 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
@@ -26,14 +26,17 @@ import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.gravitino.Config;
import org.apache.gravitino.Configs;
import org.apache.gravitino.storage.relational.JDBCBackend;
import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
+import
org.apache.gravitino.storage.relational.mapper.provider.postgresql.EntityChangeLogPostgreSQLProvider;
import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
@@ -134,11 +137,19 @@ public class TestEntityChangeLogMapper {
.findFirst()
.orElseThrow(() -> new AssertionError("recent row missing"));
- entityChangeLogMapper.pruneOldEntityChanges(1001L);
-
- List<EntityChangeRecord> after =
entityChangeLogMapper.selectEntityChanges(0L, 100);
- Assertions.assertEquals(1, after.size());
- Assertions.assertEquals(recent, after.get(0).getCreatedAt());
+ int prunedRows =
+ SessionUtils.doWithCommitAndFetchResult(
+ EntityChangeLogMapper.class,
+ mapper ->
mapper.pruneOldEntityChanges(TimeUnit.DAYS.toMillis(30)));
+ Assertions.assertEquals(1, prunedRows);
+
+ try (SqlSession verificationSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true))
{
+ List<EntityChangeRecord> after =
+
verificationSession.getMapper(EntityChangeLogMapper.class).selectEntityChanges(0L,
100);
+ Assertions.assertEquals(1, after.size());
+ Assertions.assertEquals(recent, after.get(0).getCreatedAt());
+ }
}
@Test
@@ -156,6 +167,36 @@ public class TestEntityChangeLogMapper {
Assertions.assertTrue(rows.get(1).getId() < rows.get(2).getId());
}
+ @Test
+ void testEntityChangeLogPruneUsesDatabaseTime() {
+ String baseSql =
+ new
EntityChangeLogBaseSQLProvider().pruneOldEntityChanges(TimeUnit.DAYS.toMillis(30));
+ String postgreSql =
+ new
EntityChangeLogPostgreSQLProvider().pruneOldEntityChanges(TimeUnit.DAYS.toMillis(30));
+
+ // The cutoff must be derived from the DB clock, never from a JVM
timestamp bound into the SQL.
+ Assertions.assertTrue(baseSql.contains("CURRENT_TIMESTAMP"), baseSql);
+ Assertions.assertTrue(baseSql.contains("#{retentionMs}"), baseSql);
+ Assertions.assertFalse(baseSql.contains("#{before}"), baseSql);
+ Assertions.assertTrue(postgreSql.contains("CURRENT_TIMESTAMP"),
postgreSql);
+ Assertions.assertTrue(postgreSql.contains("#{retentionMs}"), postgreSql);
+ Assertions.assertFalse(postgreSql.contains("#{before}"), postgreSql);
+ }
+
+ @Test
+ void testEntityChangeLogPruneKeepsRowsInsideRetention() {
+ entityChangeLogMapper.insertEntityChange(
+ "metalake1", "TABLE", "metalake1.cat.schema.recent",
OperateType.ALTER);
+
+ int prunedRows =
+ SessionUtils.doWithCommitAndFetchResult(
+ EntityChangeLogMapper.class,
+ mapper ->
mapper.pruneOldEntityChanges(TimeUnit.DAYS.toMillis(30)));
+
+ Assertions.assertEquals(0, prunedRows);
+ Assertions.assertEquals(1, entityChangeLogMapper.selectEntityChanges(0L,
100).size());
+ }
+
private void forceCreatedAt(String fullName, long createdAt) throws
SQLException {
try (PreparedStatement statement =
sharedSession
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 593e42f5ad..f5e28244ba 100644
--- a/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
+++ b/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
@@ -20,6 +20,8 @@ package org.apache.gravitino.tag;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -155,6 +157,8 @@ public class TestTagManager {
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_MAX_RETRIES)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_LISTENER_FAILURE_ACTION)).thenReturn("SKIP");
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index 242ac37dee..3490510f0e 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -58,8 +58,10 @@ The following table lists the storage configuration items:
| `gravitino.entity.store.deleteAfterTimeMs` | The maximum time in
milliseconds that deleted and old-version data is kept. Set to at least 10
minutes and no longer than 30 days.
|
`604800000`(7 days) | No
| 0.5.0 |
| `gravitino.entity.store.versionRetentionCount` | The Count of versions
allowed to be retained, including the current version, used to delete old
versions data. Set to at least 1 and no greater than 10.
| `1`
| No |
0.5.0 |
| `gravitino.entityChangeLog.pollIntervalSecs` | The interval in seconds
for polling the entity change log. The poller invalidates stale local caches
(e.g. the catalog cache) across HA nodes by consuming change log records. Must
be positive. | `3`
| No |
1.3.0 |
-| `gravitino.entityChangeLog.retentionSecs` | The retention time in
seconds for entity change log rows. Expired rows are pruned periodically. Set
to `0` to disable automatic cleanup. Must be non-negative.
| `86400`(1
day) | No |
1.3.0 |
-| `gravitino.entityChangeLog.cleanupIntervalSecs` | The interval in seconds
for pruning expired entity change log rows. Must be positive.
| `3600`(1
hour) | No |
1.3.0 |
+| `gravitino.entityChangeLog.listenerMaxRetries` | The number of times the
poller retries a change log batch for a failing listener before applying
`gravitino.entityChangeLog.listenerFailureAction`. Must be non-negative.
| `10`
| No |
2.0.0 |
+| `gravitino.entityChangeLog.listenerFailureAction` | What the poller does
once a listener exhausted its retries. `EXIT` stops this server, because its
local caches are known to be stale and no longer safe to serve from; `SKIP`
drops the batch for that listener and keeps serving. |
`EXIT` | No
| 2.0.0 |
+| `gravitino.entityChangeLog.retentionSecs` | The retention time in
seconds for entity change log rows. A dedicated cleaner removes rows older than
this period using database time. Set to `0` to disable cleanup, otherwise at
least 10x `pollIntervalSecs`. |
`2592000`(30 days) | No
| 1.3.0 |
+| `gravitino.entityChangeLog.cleanupIntervalSecs` | The interval in seconds
for running the dedicated entity change log cleaner. Must be positive.
| `86400`(1
day) | No |
1.3.0 |
| `gravitino.entity.store.relational` | Detailed implementation
of Relational storage. `H2`, `MySQL` and `PostgreSQL` is supported, and the
implementation is `JDBCBackend`.
|
`JDBCBackend` | No
| 0.5.0 |
| `gravitino.entity.store.relational.jdbcUrl` | The database url that
the `JDBCBackend` needs to connect to. If you use `MySQL` or `PostgreSQL`, you
should firstly initialize the database tables yourself by executing the ddl
scripts in the `${GRAVITINO_HOME}/scripts/{DATABASE_TYPE}/` directory. |
`jdbc:h2` | No
| 0.5.0 |
| `gravitino.entity.store.relational.jdbcDriver` | The jdbc driver name
that the `JDBCBackend` needs to use. You should place the driver Jar package in
the `${GRAVITINO_HOME}/libs/` directory.
|
`org.h2.Driver` | Yes if the jdbc connection url is not
`jdbc:h2` | 0.5.0 |
