This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 51fb2d29c7 [#13581] improvement(core): Drain change-log backlog
without waiting the poll interval after a full batch (#13619)
51fb2d29c7 is described below
commit 51fb2d29c7573a2284cb4f5422e5ae4fb380a989
Author: Qi Yu <[email protected]>
AuthorDate: Fri Oct 9 15:45:22 2026 +0800
[#13581] improvement(core): Drain change-log backlog without waiting the
poll interval after a full batch (#13619)
### What changes were proposed in this pull request?
- Schedule each change-log poll cycle individually: a full batch
schedules the next cycle immediately; a partial or empty batch, or a
failed cycle, waits the configured interval.
- Keep each cycle bounded by `gravitino.entityChangeLog.pollBatchSize`
(default `2000`). Sustained backlog intentionally drains continuously,
with no minimum delay. Tail diagnostics sample at most once per interval
during full batches, including failed sampling attempts; partial and
empty batches refresh the tail normally.
- On shutdown, discard future delayed polls. Already-due tasks may still
execute, so the `closed` guard prevents them from polling the closing
store. Reject restarting a closed poller.
- Synchronize configuration, metrics, and multi-node design
documentation, including database pressure and sampled-lag limitations.
### Why are the changes needed?
The poller previously read at most 2,000 records and then always slept
for the full interval. With the default three-second interval,
throughput was capped at about 667 records/s even with a backlog,
allowing remote caches to remain stale under sustained writes.
Fix: #13581
### Does this PR introduce _any_ user-facing change?
- New server property `gravitino.entityChangeLog.pollBatchSize`.
- Servers behind the change log drain continuously. Idle and failing
pollers retain the configured interval. Full batches indicate possible
backlog; zero sampled record lag alone does not prove the poller has
caught up.
### How was this patch tested?
- 22 `TestEntityChangeLogPoller` cases cover scheduling, bounded
batches, tail-query frequency and failed sampling attempts, scheduling
rejection, future-task cancellation, and an already-due task queued
during shutdown. The last test uses the production scheduler and proves
the queued task executes without polling after closure.
- H2 cross-node invalidation verifies that five changed cached tables
drain over batches of two.
- `core:test` (2,531 tests, two skipped) and all non-Docker tests in
fileset (186), Kafka (15), model (35), and generic lakehouse (96) passed
with zero failures/errors. `:core:spotlessCheck`, `:core:javadoc`, and
`git diff --check` passed. UTF-8 locale and explicit
`SKIP_DOCKER_TESTS=true` were used. The three cross-node invalidation
cases also passed on H2, MySQL, and PostgreSQL (nine backend
invocations, zero failures/errors/skips).
- Six isolated mutations were caught by the corresponding assertions:
fixed-interval draining, missing backlog signal, excessive
successful/failed tail queries, missing closed guard, and missing
delayed-task shutdown policy. A JDK 17 probe confirms future delayed
tasks are canceled while already-due tasks can run after shutdown.
---
.../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 | 17 ++
.../storage/relational/EntityChangeLogPoller.java | 186 ++++++++++---
.../storage/relational/RelationalEntityStore.java | 4 +-
.../authorization/TestAccessControlManager.java | 4 +
.../gravitino/authorization/TestOwnerManager.java | 4 +
.../gravitino/hook/TestFilesetHookDispatcher.java | 4 +
.../apache/gravitino/policy/TestPolicyManager.java | 4 +
.../gravitino/stats/TestStatisticManager.java | 4 +
.../storage/AbstractEntityStorageTest.java | 4 +
.../TestEntityCacheCrossNodeInvalidation.java | 61 ++++-
.../relational/TestEntityChangeLogPoller.java | 304 ++++++++++++++++++++-
...TestRelationalEntityStoreHierarchicalCache.java | 2 +
.../org/apache/gravitino/tag/TestTagManager.java | 4 +
.../gravitino-entity-cache-multinode-design.md | 2 +-
docs/gravitino-server-config.md | 15 +-
docs/metrics.md | 15 +-
21 files changed, 600 insertions(+), 54 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 6a9c49dfdd..3187f22b70 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
@@ -18,8 +18,10 @@
*/
package org.apache.gravitino.catalog.fileset;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
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;
@@ -278,6 +280,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 a974515386..727bf38882 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
@@ -19,8 +19,10 @@
package org.apache.gravitino.catalog.kafka;
import static org.apache.gravitino.Catalog.Type.MESSAGING;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
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;
@@ -167,6 +169,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 6bed860b0b..df25b219c6 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
@@ -18,8 +18,10 @@
*/
package org.apache.gravitino.catalog.lakehouse.generic;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
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;
@@ -126,6 +128,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 4ef7a9f3be..0793521deb 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
@@ -18,8 +18,10 @@
*/
package org.apache.gravtitino.catalog.model;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -129,6 +131,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 16e8eed9c5..c78aacde16 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
@@ -19,9 +19,11 @@
package org.apache.gravitino.cache.it;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
import static
org.apache.gravitino.Configs.DEFAULT_RELATIONAL_JDBC_BACKEND_MAX_IDLE_CONNECTIONS;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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;
@@ -298,6 +300,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 108f316219..600d5da8e5 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -205,6 +205,9 @@ public class Configs {
.createWithDefault(60 * 60 * 1000L);
public static final long DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS = 3L;
+ /** Default maximum number of entity change log records read in one polling
cycle. */
+ public static final int DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE = 2000;
+
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;
@@ -216,6 +219,20 @@ public class Configs {
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS);
+ /**
+ * Maximum records read per entity change log poll. A full batch triggers an
immediate next poll;
+ * empty or partial batches wait for the configured poll interval.
+ */
+ public static final ConfigEntry<Integer> ENTITY_CHANGE_LOG_POLL_BATCH_SIZE =
+ new ConfigBuilder("gravitino.entityChangeLog.pollBatchSize")
+ .doc(
+ "The maximum number of entity change log records read per poll.
A full batch is"
+ + " followed by another poll right away instead of after the
poll interval")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
+
public static final ConfigEntry<Long> ENTITY_CHANGE_LOG_RETENTION_SECS =
new ConfigBuilder("gravitino.entityChangeLog.retentionSecs")
.doc("The retention time in seconds for entity change logs. Set 0 to
disable cleanup")
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java
index b42b57dc1a..9624a4b9f3 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
@@ -23,10 +23,12 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
+import org.apache.gravitino.Configs;
import org.apache.gravitino.metrics.source.EntityChangeLogMetricsSource;
import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
@@ -64,29 +66,35 @@ import org.slf4j.LoggerFactory;
*
* <p>Every listener failure is logged at {@code ERROR}, so a listener that
keeps failing stays
* visible in the logs even though the poller keeps going.
+ *
+ * <p>Each poll cycle reads at most one batch. A full batch means more rows
may be waiting, so the
+ * next cycle starts right away instead of after the poll interval; an empty
or partial batch, or a
+ * failed cycle, waits the full interval. A backlog is therefore drained as
fast as batches can be
+ * read and delivered, while an idle or failing poller still polls only once
per interval.
*/
public class EntityChangeLogPoller implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(EntityChangeLogPoller.class);
- /** Max entity-change rows to fetch per batch. */
- private static final int ENTITY_CHANGE_POLLER_MAX_ROWS = 2000;
-
/** Max records rendered in a batch summary log line. */
private static final int MAX_SUMMARIZED_RECORDS = 20;
private final List<EntityChangeLogListener> listeners = new
CopyOnWriteArrayList<>();
private final long pollIntervalSecs;
+ private final int batchSize;
private final EntityChangeLogMetricsSource metrics;
private ScheduledExecutorService scheduler;
+ private volatile boolean closed = false;
private volatile long entityPollHighWaterId = 0;
+ private boolean tailSampleAttempted;
+ private long lastTailSampleNanos;
/**
* Creates an {@link EntityChangeLogPoller} with an unregistered metrics
source for callers that
* do not use the server metrics system.
*
- * @param pollIntervalSecs interval between successive polling cycles
+ * @param pollIntervalSecs interval between polling cycles once the poller
has caught up
*/
public EntityChangeLogPoller(long pollIntervalSecs) {
this(pollIntervalSecs, new EntityChangeLogMetricsSource());
@@ -95,13 +103,28 @@ public class EntityChangeLogPoller implements
AutoCloseable {
/**
* Creates a poller using the metrics source registered by the entity store.
*
- * @param pollIntervalSecs interval between successive polling cycles
+ * @param pollIntervalSecs interval between polling cycles once the poller
has caught up
* @param metrics process-local change log metrics
*/
public EntityChangeLogPoller(long pollIntervalSecs,
EntityChangeLogMetricsSource metrics) {
+ this(pollIntervalSecs, Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE,
metrics);
+ }
+
+ /**
+ * Creates a poller with a custom batch size, using the metrics source
registered by the entity
+ * store.
+ *
+ * @param pollIntervalSecs interval between polling cycles once the poller
has caught up
+ * @param batchSize maximum number of change log rows read per polling cycle
+ * @param metrics process-local change log metrics
+ */
+ public EntityChangeLogPoller(
+ long pollIntervalSecs, int batchSize, EntityChangeLogMetricsSource
metrics) {
Preconditions.checkArgument(pollIntervalSecs > 0, "pollIntervalSecs must
be positive");
+ Preconditions.checkArgument(batchSize > 0, "batchSize must be positive");
this.metrics = Preconditions.checkNotNull(metrics, "metrics cannot be
null");
this.pollIntervalSecs = pollIntervalSecs;
+ this.batchSize = batchSize;
}
/**
@@ -136,7 +159,8 @@ public class EntityChangeLogPoller implements AutoCloseable
{
}
/**
- * Initializes the high-water cursor to the current DB tail and schedules
periodic polling.
+ * Initializes the high-water cursor to the current DB tail and schedules
the first poll one
+ * interval later. A poller cannot be started again after {@link #close()}.
*
* <p>On every start (including restarts), the cursor is set to the current
maximum change ID in
* the DB, so historical change records written before this server process
started are NOT
@@ -145,33 +169,30 @@ public class EntityChangeLogPoller implements
AutoCloseable {
* cache.
*/
public void start() {
+ Preconditions.checkState(!closed, "A closed entity change log poller
cannot be restarted");
entityPollHighWaterId =
getOrDefault(
SessionUtils.getWithoutCommit(
EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
metrics.setDbTailId(entityPollHighWaterId);
+ tailSampleAttempted = true;
+ lastTailSampleNanos = nanoTime();
metrics.setCursorId(entityPollHighWaterId);
LOG.info(
- "Starting entity change log poller at high-water id {} with a {}
second interval, "
- + "{} listener(s) registered",
+ "Starting entity change log poller at high-water id {} with a {}
second interval and a "
+ + "batch size of {}, {} listener(s) registered",
entityPollHighWaterId,
pollIntervalSecs,
+ batchSize,
listeners.size());
- scheduler =
- Executors.newSingleThreadScheduledExecutor(
- r -> {
- Thread t = new Thread(r);
- t.setName("Gravitino-EntityChangeLogPoller");
- t.setDaemon(true);
- return t;
- });
- scheduler.scheduleWithFixedDelay(
- this::pollChanges, pollIntervalSecs, pollIntervalSecs,
TimeUnit.SECONDS);
+ scheduler = createScheduler();
+ scheduleNextPoll(TimeUnit.SECONDS.toMillis(pollIntervalSecs));
}
@Override
public void close() {
+ closed = true;
if (scheduler != null) {
scheduler.shutdown();
try {
@@ -190,28 +211,99 @@ public class EntityChangeLogPoller implements
AutoCloseable {
}
@VisibleForTesting
- void pollChanges() {
+ ScheduledExecutorService createScheduler() {
+ ScheduledThreadPoolExecutor executor =
+ new ScheduledThreadPoolExecutor(
+ 1,
+ r -> {
+ Thread t = new Thread(r);
+ t.setName("Gravitino-EntityChangeLogPoller");
+ t.setDaemon(true);
+ return t;
+ });
+ // Drop polls whose delay has not elapsed so close() does not wait for an
idle poll. Already-due
+ // tasks, including zero-delay drain hops, can still run after shutdown();
runPollCycle's closed
+ // guard prevents those tasks from accessing the closing store.
+ executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+ return executor;
+ }
+
+ @VisibleForTesting
+ long nanoTime() {
+ return System.nanoTime();
+ }
+
+ /**
+ * Runs one poll cycle and schedules the next one: immediately when the
cycle read a full batch,
+ * otherwise after the poll interval. A failed cycle also waits the full
interval, so a database
+ * outage is not retried in a tight loop.
+ */
+ private void runPollCycle() {
+ if (closed) {
+ return;
+ }
+ long nextDelayMillis = TimeUnit.SECONDS.toMillis(pollIntervalSecs);
+ try {
+ if (pollChanges()) {
+ // Continuous draining is intentional: every cycle advances the cursor
and is bounded by
+ // batchSize. Tail sampling is rate-limited separately so diagnostics
do not double the
+ // database query rate while a sustained backlog is being consumed.
+ nextDelayMillis = 0;
+ }
+ } finally {
+ // pollChanges() catches Throwable, but scheduling in finally keeps the
poller alive even if
+ // that ever changes: a cycle that is not rescheduled stops cache
invalidation for good.
+ scheduleNextPoll(nextDelayMillis);
+ }
+ }
+
+ private void scheduleNextPoll(long delayMillis) {
+ if (closed) {
+ return;
+ }
+ try {
+ scheduler.schedule(this::runPollCycle, delayMillis,
TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // Only a shut-down scheduler rejects work, and close() sets closed
before shutting down.
+ if (!closed) {
+ LOG.error("Could not schedule the next entity change log poll", e);
+ }
+ }
+ }
+
+ /**
+ * Polls one batch and delivers it to the listeners.
+ *
+ * @return true if the batch was full, so more rows may be waiting and the
next poll should not
+ * wait for the interval; false if the poller caught up or the poll
failed
+ */
+ @VisibleForTesting
+ boolean pollChanges() {
try (Timer.Context ignored = metrics.timePoll()) {
- doPollChanges();
+ return doPollChanges();
} catch (Throwable e) {
- // Catch Throwable, not Exception: this method is the task handed to
- // scheduleWithFixedDelay(), and anything that escapes it cancels all
future runs for good,
- // silently. A listener or its recovery path can throw an Error as well
as an Exception.
- // Losing the poller would stop cache invalidation for every listener in
this process, so we
- // log and let the next cycle run.
+ // Catch Throwable, not Exception: a listener or its recovery path can
throw an Error as well
+ // as an Exception. Losing the poller would stop cache invalidation for
every listener in
+ // this process, so we log and let the next cycle run after the normal
interval.
if (handleInterruptIfAny(e, "Entity change poll")) {
- return;
+ return false;
}
metrics.pollFailed();
LOG.warn("Entity change poll failed at high-water id {}",
entityPollHighWaterId, e);
+ return false;
}
}
- private synchronized void doPollChanges() {
+ private synchronized boolean doPollChanges() {
BatchDelivery delivery = fetchNextDelivery();
- if (delivery != null) {
- deliver(delivery);
+ if (delivery == null) {
+ return false;
}
+ deliver(delivery);
+ // A full batch is the backlog signal. The sampled tail id would be more
precise, but that
+ // sample is best-effort and may be missing; a full batch with nothing
behind it only costs one
+ // extra empty poll before the normal interval resumes.
+ return delivery.changes.size() >= batchSize;
}
@Nullable
@@ -221,18 +313,27 @@ public class EntityChangeLogPoller implements
AutoCloseable {
// The tail is for observability only. A failed sample must not suppress
delivery of rows
// already fetched successfully or hold the cursor back.
@Nullable Long dbTailId = null;
- try {
- dbTailId =
- getOrDefault(
- SessionUtils.getWithoutCommit(
- EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
- metrics.setDbTailId(dbTailId);
- } catch (RuntimeException e) {
- if (handleInterruptIfAny(e, "Entity change log tail sample")) {
- throw e;
+ long nowNanos = nanoTime();
+ if (changes.size() < batchSize
+ || !tailSampleAttempted
+ || nowNanos - lastTailSampleNanos >=
TimeUnit.SECONDS.toNanos(pollIntervalSecs)) {
+ // A full batch already signals backlog. Sample at most once per
interval during a drain,
+ // including failed attempts, while partial/empty polls refresh the
caught-up state.
+ tailSampleAttempted = true;
+ lastTailSampleNanos = nowNanos;
+ try {
+ dbTailId =
+ getOrDefault(
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
+ metrics.setDbTailId(dbTailId);
+ } catch (RuntimeException e) {
+ if (handleInterruptIfAny(e, "Entity change log tail sample")) {
+ throw e;
+ }
+ metrics.tailSampleFailed();
+ LOG.warn("Could not sample entity change log tail; retaining the
previous gauge value", e);
}
- metrics.tailSampleFailed();
- LOG.warn("Could not sample entity change log tail; retaining the
previous gauge value", e);
}
metrics.pollSucceeded(changes.size());
long durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
fetchStartNanos);
@@ -291,8 +392,7 @@ public class EntityChangeLogPoller implements AutoCloseable
{
private List<EntityChangeRecord> fetchEntityChanges() {
return SessionUtils.getWithoutCommit(
- EntityChangeLogMapper.class,
- m -> m.selectEntityChanges(entityPollHighWaterId,
ENTITY_CHANGE_POLLER_MAX_ROWS));
+ EntityChangeLogMapper.class, m ->
m.selectEntityChanges(entityPollHighWaterId, batchSize));
}
private static boolean handleInterruptIfAny(Throwable e, String context) {
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 748518fbda..3d9026739b 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
@@ -120,7 +120,9 @@ public class RelationalEntityStore
}
this.entityChangeLogPoller =
new EntityChangeLogPoller(
- config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS),
changeLogMetrics);
+ config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS),
+ config.get(Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE),
+ changeLogMetrics);
this.entityChangeLogCleaner =
new EntityChangeLogCleaner(
TimeUnit.SECONDS.toMillis(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)),
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
index 01d92dd44d..dd4310cf18 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.authorization;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
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;
@@ -156,6 +158,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 e17d9b6f10..767e13d582 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.authorization;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -116,6 +118,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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/hook/TestFilesetHookDispatcher.java
b/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
index 4037f5d2c7..c1622ab676 100644
---
a/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.hook;
import static org.apache.gravitino.Configs.CATALOG_CACHE_EVICTION_INTERVAL_MS;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
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;
@@ -248,6 +250,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 dedb7e164e..eae8c12874 100644
--- a/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
+++ b/core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.policy;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
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;
@@ -248,6 +250,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 46e82715d8..52f931890b 100644
--- a/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
+++ b/core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.stats;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -115,6 +117,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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 7c8142d4b7..896fe90ca4 100644
---
a/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
+++
b/core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.storage;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
@@ -149,6 +151,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
index 5c979bdfb2..a304b28888 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityCacheCrossNodeInvalidation.java
@@ -19,12 +19,16 @@
package org.apache.gravitino.storage.relational;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
import org.apache.gravitino.Entity.EntityType;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.cache.CaffeineEntityCache;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.metrics.source.EntityChangeLogMetricsSource;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.junit.jupiter.api.Assertions;
@@ -45,7 +49,12 @@ public class TestEntityCacheCrossNodeInvalidation extends
TestJDBCBackend {
// A large poll interval so the background scheduler never fires during the
test; the test drives
// node B's poll explicitly via pollChanges().
private EntityChangeLogPoller newIdlePoller(CaffeineEntityCache nodeBCache) {
- EntityChangeLogPoller poller = new EntityChangeLogPoller(3600);
+ return newIdlePoller(nodeBCache,
Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
+ }
+
+ private EntityChangeLogPoller newIdlePoller(CaffeineEntityCache nodeBCache,
int batchSize) {
+ EntityChangeLogPoller poller =
+ new EntityChangeLogPoller(3600, batchSize, new
EntityChangeLogMetricsSource());
poller.registerListener(new EntityCacheChangeLogListener(nodeBCache));
// start() seeds the cursor with the current DB tail, modelling a node
whose cache is already
// warm: only changes written after this point are replayed.
@@ -85,6 +94,56 @@ public class TestEntityCacheCrossNodeInvalidation extends
TestJDBCBackend {
}
}
+ @TestTemplate
+ void testBacklogLargerThanOneBatchReachesEveryDistinctEntityOnNodeB() throws
IOException {
+ createParentEntities(METALAKE, CATALOG, SCHEMA, AUDIT_INFO);
+ Namespace namespace = Namespace.of(METALAKE, CATALOG, SCHEMA);
+ List<TableEntity> tables = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ TableEntity table =
+ createTableEntity(
+ RandomIdGenerator.INSTANCE.nextId(), namespace, "backlog_table"
+ i, AUDIT_INFO);
+ backend.insert(table, false);
+ tables.add(table);
+ }
+
+ // Node B has cached every table.
+ CaffeineEntityCache nodeBCache = new CaffeineEntityCache(new Config() {});
+ tables.forEach(nodeBCache::put);
+
+ EntityChangeLogPoller nodeBPoller = newIdlePoller(nodeBCache, 2);
+ try {
+ // Node A renames every table, leaving five change rows: more than two
batches of two.
+ for (TableEntity table : tables) {
+ backend.update(
+ table.nameIdentifier(),
+ EntityType.TABLE,
+ entity ->
+ createTableEntity(table.id(), namespace, table.name() +
"_renamed", AUDIT_INFO));
+ }
+
+ // A poll reads one bounded batch and reports that more rows may be
waiting.
+ Assertions.assertTrue(nodeBPoller.pollChanges());
+ Assertions.assertEquals(3, countCached(nodeBCache, tables));
+
+ // The scheduler polls again immediately while batches are full, until
one is partial.
+ int polls = 1;
+ while (nodeBPoller.pollChanges()) {
+ polls++;
+ Assertions.assertTrue(polls < 10, "the backlog should drain in a
bounded number of polls");
+ }
+ Assertions.assertEquals(0, countCached(nodeBCache, tables));
+ } finally {
+ nodeBPoller.close();
+ }
+ }
+
+ private static long countCached(CaffeineEntityCache cache, List<TableEntity>
tables) {
+ return tables.stream()
+ .filter(table -> cache.contains(table.nameIdentifier(),
EntityType.TABLE))
+ .count();
+ }
+
@TestTemplate
void testSchemaDropOnNodeAClearsChildTablesOnNodeB() throws IOException {
createParentEntities(METALAKE, CATALOG, SCHEMA, AUDIT_INFO);
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 2d68d14d68..2b086f2f61 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
@@ -18,16 +18,37 @@
*/
package org.apache.gravitino.storage.relational;
+import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.spy;
+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.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
+import java.util.stream.Collectors;
import org.apache.gravitino.metrics.source.EntityChangeLogMetricsSource;
import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper;
import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
@@ -35,6 +56,8 @@ import
org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
import org.mockito.MockedStatic;
public class TestEntityChangeLogPoller {
@@ -45,6 +68,190 @@ public class TestEntityChangeLogPoller {
void testRejectsInvalidConfiguration() {
Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(0));
Assertions.assertThrows(IllegalArgumentException.class, () -> new
EntityChangeLogPoller(-1));
+ EntityChangeLogMetricsSource metrics = new EntityChangeLogMetricsSource();
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> new EntityChangeLogPoller(1, 0,
metrics));
+ }
+
+ @Test
+ void testPollChangesReportsBacklogOnlyForFullBatch() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 2)).thenReturn(changes(1L, 2L));
+ when(mapper.selectEntityChanges(2L, 2)).thenReturn(changes(3L, 3L));
+ when(mapper.selectEntityChanges(3L, 2)).thenReturn(List.of());
+ List<EntityChangeRecord> received = new ArrayList<>();
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogMetricsSource metrics = new
EntityChangeLogMetricsSource();
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(1, 2, metrics);
+ poller.registerListener(received::addAll);
+
+ Assertions.assertTrue(poller.pollChanges(), "a full batch may leave a
backlog");
+ Assertions.assertFalse(poller.pollChanges(), "a partial batch means the
poller caught up");
+ Assertions.assertFalse(poller.pollChanges(), "an empty poll means the
poller caught up");
+ Assertions.assertEquals(
+ 3L,
metrics.getMetricRegistry().getGauges().get("cursor-id").getValue());
+ }
+
+ Assertions.assertEquals(List.of(1L, 2L, 3L), ids(received));
+ }
+
+ @Test
+ void testFullBatchSchedulesNextPollWithoutWaitingForTheInterval() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 2)).thenReturn(changes(1L, 2L));
+ when(mapper.selectEntityChanges(2L, 2)).thenReturn(changes(3L, 3L));
+ when(mapper.selectEntityChanges(3L, 2)).thenThrow(new
RuntimeException("database down"));
+ ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
+ ArgumentCaptor<Runnable> cycle = ArgumentCaptor.forClass(Runnable.class);
+ List<EntityChangeRecord> received = new ArrayList<>();
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+
+ EntityChangeLogPoller poller =
+ spy(new EntityChangeLogPoller(5, 2, new
EntityChangeLogMetricsSource()));
+ doReturn(scheduler).when(poller).createScheduler();
+ poller.registerListener(received::addAll);
+
+ poller.start();
+ InOrder schedules = inOrder(scheduler);
+ schedules.verify(scheduler).schedule(cycle.capture(), eq(5000L),
eq(TimeUnit.MILLISECONDS));
+
+ // A full batch: more rows may be waiting, so the next cycle starts
right away.
+ cycle.getValue().run();
+ schedules.verify(scheduler).schedule(any(Runnable.class), eq(0L),
eq(TimeUnit.MILLISECONDS));
+
+ // A partial batch: caught up, so wait the normal interval.
+ cycle.getValue().run();
+ schedules
+ .verify(scheduler)
+ .schedule(any(Runnable.class), eq(5000L), eq(TimeUnit.MILLISECONDS));
+
+ // A failed poll keeps the normal interval instead of retrying in a
tight loop.
+ cycle.getValue().run();
+ schedules
+ .verify(scheduler)
+ .schedule(any(Runnable.class), eq(5000L), eq(TimeUnit.MILLISECONDS));
+
+ // Once closed, a cycle that is still running does not schedule another
one.
+ poller.close();
+ cycle.getValue().run();
+ verify(scheduler, times(4)).schedule(any(Runnable.class), anyLong(),
any());
+ }
+
+ Assertions.assertEquals(List.of(1L, 2L, 3L), ids(received));
+ }
+
+ @Test
+ void testDrainSamplesTailOncePerIntervalAndRefreshesAfterCatchingUp() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 2)).thenReturn(changes(1L, 2L));
+ when(mapper.selectEntityChanges(2L, 2)).thenReturn(changes(3L, 4L));
+ when(mapper.selectEntityChanges(4L, 2)).thenReturn(changes(5L, 6L));
+ when(mapper.selectEntityChanges(6L, 2)).thenReturn(changes(7L, 7L));
+ when(mapper.selectEntityChanges(7L, 2)).thenReturn(List.of());
+ when(mapper.selectMaxChangeId()).thenReturn(10L, 12L, 13L, 14L);
+ AtomicLong clock = new AtomicLong();
+ List<EntityChangeRecord> received = new ArrayList<>();
+ EntityChangeLogMetricsSource metrics = new EntityChangeLogMetricsSource();
+ EntityChangeLogPoller poller = spy(new EntityChangeLogPoller(5, 2,
metrics));
+ doAnswer(invocation -> clock.get()).when(poller).nanoTime();
+ poller.registerListener(received::addAll);
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ Assertions.assertTrue(poller.pollChanges());
+ clock.set(TimeUnit.SECONDS.toNanos(1));
+ Assertions.assertTrue(poller.pollChanges());
+ verify(mapper, times(1)).selectMaxChangeId();
+ Assertions.assertEquals(
+ 10L,
metrics.getMetricRegistry().getGauges().get("db-tail-id").getValue());
+
+ clock.set(TimeUnit.SECONDS.toNanos(5));
+ Assertions.assertTrue(poller.pollChanges());
+ verify(mapper, times(2)).selectMaxChangeId();
+ Assertions.assertFalse(poller.pollChanges());
+ Assertions.assertFalse(poller.pollChanges());
+ verify(mapper, times(4)).selectMaxChangeId();
+ }
+ Assertions.assertEquals(List.of(1L, 2L, 3L, 4L, 5L, 6L, 7L),
ids(received));
+ Assertions.assertEquals(
+ 7L,
metrics.getMetricRegistry().getGauges().get("cursor-id").getValue());
+ Assertions.assertEquals(
+ 14L,
metrics.getMetricRegistry().getGauges().get("db-tail-id").getValue());
+ }
+
+ @Test
+ void testFailedTailSamplesAreAlsoRateLimitedDuringDrain() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ when(mapper.selectEntityChanges(0L, 2)).thenReturn(changes(1L, 2L));
+ when(mapper.selectEntityChanges(2L, 2)).thenReturn(changes(3L, 4L));
+ when(mapper.selectEntityChanges(4L, 2)).thenReturn(changes(5L, 6L));
+ when(mapper.selectMaxChangeId()).thenThrow(new RuntimeException("tail
unavailable"));
+ AtomicLong clock = new AtomicLong();
+ List<EntityChangeRecord> received = new ArrayList<>();
+ EntityChangeLogMetricsSource metrics = new EntityChangeLogMetricsSource();
+ EntityChangeLogPoller poller = spy(new EntityChangeLogPoller(5, 2,
metrics));
+ doAnswer(invocation -> clock.get()).when(poller).nanoTime();
+ poller.registerListener(received::addAll);
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ Assertions.assertTrue(poller.pollChanges());
+ clock.set(TimeUnit.SECONDS.toNanos(1));
+ Assertions.assertTrue(poller.pollChanges());
+ verify(mapper, times(1)).selectMaxChangeId();
+ clock.set(TimeUnit.SECONDS.toNanos(5));
+ Assertions.assertTrue(poller.pollChanges());
+ verify(mapper, times(2)).selectMaxChangeId();
+ }
+ Assertions.assertEquals(List.of(1L, 2L, 3L, 4L, 5L, 6L), ids(received));
+ Assertions.assertEquals(
+ 6L,
metrics.getMetricRegistry().getGauges().get("cursor-id").getValue());
+ Assertions.assertEquals(
+ 2,
metrics.getMetricRegistry().counter("tail-sample-failures-total").getCount());
+ Assertions.assertEquals(
+ 0,
metrics.getMetricRegistry().counter("poll-failures-total").getCount());
+ }
+
+ @Test
+ void testSchedulingRejectionDoesNotEscapeStart() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
+ doThrow(new RejectedExecutionException("scheduler stopped"))
+ .when(scheduler)
+ .schedule(any(Runnable.class), anyLong(), any());
+ EntityChangeLogPoller poller = spy(new EntityChangeLogPoller(1));
+ doReturn(scheduler).when(poller).createScheduler();
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ Assertions.assertDoesNotThrow(poller::start);
+ verify(scheduler).schedule(any(Runnable.class), anyLong(), any());
+ poller.close();
+ }
+ }
+
+ @Test
+ void testSchedulingRejectionWhileClosingDoesNotEscapeStart() {
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
+ EntityChangeLogPoller poller = spy(new EntityChangeLogPoller(1));
+ doReturn(scheduler).when(poller).createScheduler();
+ doAnswer(
+ invocation -> {
+ poller.close();
+ throw new RejectedExecutionException("closed during scheduling");
+ })
+ .when(scheduler)
+ .schedule(any(Runnable.class), anyLong(), any());
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ Assertions.assertDoesNotThrow(poller::start);
+ verify(scheduler).shutdown();
+ verify(scheduler).schedule(any(Runnable.class), anyLong(), any());
+ }
}
@Test
@@ -132,7 +339,7 @@ public class TestEntityChangeLogPoller {
EntityChangeLogPoller poller = new EntityChangeLogPoller(1);
// A listener that clears its catalog cache can close an
IsolatedClassLoader that is still in
// use, and a request holding a class from it then fails with
NoClassDefFoundError, an Error
- // rather than an Exception. pollChanges() is the task given to
scheduleWithFixedDelay(), so
+ // rather than an Exception. pollChanges() runs inside each scheduled
cycle, so
// anything escaping it would cancel every future poll and freeze
invalidation process-wide.
poller.registerListener(
changes -> {
@@ -389,6 +596,97 @@ public class TestEntityChangeLogPoller {
0,
metrics.getMetricRegistry().counter("records-applied-total").getCount());
}
+ @Test
+ void testCloseDropsThePendingPollInsteadOfWaitingForIt() {
+ // Uses the real scheduler, with the pending poll two seconds away. If
that delayed task
+ // survived
+ // shutdown, close() would wait for it. No poll may run after close()
either. A poll would run
+ // on the scheduler thread, where this thread's static SessionUtils mock
is not active, so it is
+ // observed through the poll timer, which counts every attempt, failed or
not.
+ EntityChangeLogMetricsSource metrics = new EntityChangeLogMetricsSource();
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ long closeMillis;
+
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ EntityChangeLogPoller poller = new EntityChangeLogPoller(2, metrics);
+ poller.start();
+ long startNanos = System.nanoTime();
+ poller.close();
+ closeMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNanos);
+ Assertions.assertThrows(IllegalStateException.class, poller::start);
+ }
+
+ Assertions.assertTrue(closeMillis < 1500, "close() took " + closeMillis +
" ms");
+ Assertions.assertEquals(0,
metrics.getMetricRegistry().timer("poll-duration").getCount());
+ }
+
+ @Test
+ void testAlreadyDueDrainHopDoesNotPollAfterClose() throws Exception {
+ EntityChangeLogMetricsSource metrics = new EntityChangeLogMetricsSource();
+ EntityChangeLogPoller poller = spy(new EntityChangeLogPoller(1, metrics));
+ ScheduledThreadPoolExecutor executor = (ScheduledThreadPoolExecutor)
poller.createScheduler();
+ ScheduledExecutorService scheduler =
+ mock(ScheduledExecutorService.class, delegatesTo(executor));
+ doReturn(scheduler).when(poller).createScheduler();
+ CountDownLatch workerBusy = new CountDownLatch(1);
+ CountDownLatch releaseWorker = new CountDownLatch(1);
+ CountDownLatch shutdownStarted = new CountDownLatch(1);
+ AtomicReference<ScheduledFuture<?>> queuedHop = new AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ // Queue the real poll-cycle runnable as an already-due drain
hop behind a busy
+ // worker.
+ ScheduledFuture<?> hop =
+ executor.schedule(invocation.<Runnable>getArgument(0), 0,
TimeUnit.MILLISECONDS);
+ queuedHop.set(hop);
+ return hop;
+ })
+ .when(scheduler)
+ .schedule(any(Runnable.class), anyLong(), any());
+ doAnswer(
+ invocation -> {
+ executor.shutdown();
+ shutdownStarted.countDown();
+ return null;
+ })
+ .when(scheduler)
+ .shutdown();
+ ExecutorService closer = Executors.newSingleThreadExecutor();
+ Future<?> blockingTask =
+ executor.submit(
+ () -> {
+ workerBusy.countDown();
+ try {
+ Assertions.assertTrue(releaseWorker.await(5,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ });
+ try {
+ Assertions.assertTrue(workerBusy.await(5, TimeUnit.SECONDS));
+ EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
+ try (MockedStatic<SessionUtils> sessionUtils =
mockStatic(SessionUtils.class)) {
+ mockSessionUtils(sessionUtils, mapper);
+ poller.start();
+ }
+ Future<?> closed = closer.submit(poller::close);
+ Assertions.assertTrue(shutdownStarted.await(5, TimeUnit.SECONDS));
+ releaseWorker.countDown();
+ blockingTask.get(5, TimeUnit.SECONDS);
+ // The shutdown policy retains already-due tasks: get() proves this hop
ran, not cancelled.
+ queuedHop.get().get(5, TimeUnit.SECONDS);
+ closed.get(5, TimeUnit.SECONDS);
+ Assertions.assertEquals(0,
metrics.getMetricRegistry().timer("poll-duration").getCount());
+ } finally {
+ releaseWorker.countDown();
+ executor.shutdownNow();
+ closer.shutdownNow();
+ poller.close();
+ }
+ }
+
@Test
void testDispatchesImmutableBatchToListeners() {
EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class);
@@ -425,6 +723,10 @@ public class TestEntityChangeLogPoller {
return changes;
}
+ private static List<Long> ids(List<EntityChangeRecord> changes) {
+ return
changes.stream().map(EntityChangeRecord::getId).collect(Collectors.toList());
+ }
+
private static void mockSessionUtils(
MockedStatic<SessionUtils> sessionUtils, EntityChangeLogMapper mapper) {
sessionUtils
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
index 9fa204a8d2..00197d0c6d 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
@@ -205,6 +205,8 @@ public class TestRelationalEntityStoreHierarchicalCache {
Mockito.when(config.get(Configs.STORE_DELETE_AFTER_TIME)).thenReturn(20 *
60 * 1000L);
Mockito.when(config.get(Configs.VERSION_RETENTION_COUNT)).thenReturn(1L);
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
+ .thenReturn(Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24
* 60 * 60L);
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
diff --git a/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
b/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
index de3f9c8473..31ea975306 100644
--- a/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
+++ b/core/src/test/java/org/apache/gravitino/tag/TestTagManager.java
@@ -18,8 +18,10 @@
*/
package org.apache.gravitino.tag;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
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_POLL_BATCH_SIZE;
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;
@@ -173,6 +175,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_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
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/design-docs/gravitino-entity-cache-multinode-design.md
b/design-docs/gravitino-entity-cache-multinode-design.md
index a4dc346642..4d0bedb14c 100644
--- a/design-docs/gravitino-entity-cache-multinode-design.md
+++ b/design-docs/gravitino-entity-cache-multinode-design.md
@@ -338,7 +338,7 @@ After this, everything a per-node cache serves is safe or
bounded by change-log
For everything that stays in the cache:
- The node that made the change sees it right away.
-- Under normal operation, every other node sees it after the next successful
poll. The interval is set by `gravitino.entityChangeLog.pollIntervalSecs`
(**default 3 seconds**; lower it, e.g. to 1 second, for a shorter delay at the
cost of more frequent DB polls). Database failures or a listener that fails to
recover can extend this delay.
+- Under normal operation, every other node sees it once a successful poll
reaches the change. Each poll reads at most
`gravitino.entityChangeLog.pollBatchSize` records (**default 2000**); a full
batch schedules the next poll immediately to drain backlog. After an empty or
partial batch, or a failed poll, the interval is set by
`gravitino.entityChangeLog.pollIntervalSecs` (**default 3 seconds**).
Propagation includes the time to drain earlier records, and database or
listener recovery fai [...]
- The cache's own TTL (minutes to hours) is only a safety net in case the
poller ever misses a row; it is not the main mechanism.
The poller uses gap-safe, id-based polling (see `#11736`). It delivers a batch
once and advances its shared cursor even when a listener fails, so each
listener must recover locally by clearing its cache. The cache TTL remains a
backstop. Operators can use the [entity change log
metrics](../docs/metrics.md#entity-change-log-metrics) to watch the sampled
database tail, cursor, record lag, last successful poll age, listener failures,
and fallback clears; the debug logs correlate the write, [...]
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index 672dd01340..44ca811a29 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -327,9 +327,22 @@ Caches are local to each server, so a metalake modified on
one server would othe
its neighbors. Every server writes its changes to an entity change log table
and polls that table
to invalidate what other servers have touched. A separate cleaner trims old
rows.
+Each poll reads at most `pollBatchSize` records. When a poll returns a full
batch, the next poll
+starts immediately instead of waiting `pollIntervalSecs`, so a backlog is
drained as fast as the
+server can read and apply batches rather than at a fixed `pollBatchSize /
pollIntervalSecs`
+records per second. Once a poll returns fewer records than the batch size, or
fails, the server
+waits `pollIntervalSecs` again. A change written on one server therefore
normally becomes visible
+on the others within about `pollIntervalSecs`, plus the time needed to drain
any backlog ahead of
+it. Continuous draining deliberately has no minimum delay between batches; a
sustained backlog
+can therefore keep the data query running at the rate the database and
listeners support. The
+observability-only database-tail query is sampled at most once per
`pollIntervalSecs` during a
+full-batch drain; partial and empty polls still sample the tail. Each poll
holds its whole batch
+in memory, so raise `pollBatchSize` in moderate steps.
+
| Configuration Item | Description
| Default Value |
|-------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|---------------------|
-| `gravitino.entityChangeLog.pollIntervalSecs` | Interval in seconds
between polls. Must be positive.
| `3` |
+| `gravitino.entityChangeLog.pollIntervalSecs` | Interval in seconds
between polls once a server has caught up. Must be positive.
| `3` |
+| `gravitino.entityChangeLog.pollBatchSize` | Maximum number of change
log records read per poll. A full batch is followed by another poll right away.
Must be positive. | `2000` |
| `gravitino.entityChangeLog.retentionSecs` | How long in seconds change
log rows are kept, measured by database time. `0` disables cleanup; otherwise
use at least ten times `pollIntervalSecs`. | `2592000` (30 days) |
| `gravitino.entityChangeLog.cleanupIntervalSecs` | Interval in seconds
between cleaner runs. Must be positive.
| `86400` (1 day) |
diff --git a/docs/metrics.md b/docs/metrics.md
index b7e19b32c9..5d54df52f5 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -77,8 +77,9 @@
gravitino_catalog_datasource_max_connections{provider="jdbc",metalake="test_meta
The `entity-change-log` source exposes each server's change-log processing
state through JMX and
`/prometheus/metrics`. For example, `entity-change-log.record-lag` in the
metrics registry becomes
`entity_change_log_record_lag` in Prometheus. Gauges read only in-memory
values; the poller samples
-the database tail once per cycle. If only the tail sample fails, delivery
continues and the tail
-value remains at its last successful sample.
+the database tail on partial and empty polls, and at most once per
`pollIntervalSecs` during
+a full-batch drain (including failed sample attempts). If only the tail sample
fails, delivery
+continues and the tail value remains at its last successful sample.
| Metric suffix | Type and unit
| Meaning
|
| ------------------------------------------------------------ |
------------------------ |
-----------------------------------------------------------------------------------------------------------------------------------------------------------
|
@@ -92,7 +93,7 @@ value remains at its last successful sample.
| `records-fetched-total`, `records-delivered-total` | counter,
records | Rows fetched and rows delivered successfully to listeners;
one row delivered to two listeners counts twice as delivered.
|
| `records-delivered.<class>-total` | counter,
records | Successful deliveries by listener class. Lambda and anonymous
listeners share the `anonymous` bucket.
|
| `records-applied-total` | counter,
invalidations | Targeted entity-cache invalidations completed successfully;
malformed rows and fallback clears do not count.
|
-| `batch-size-records` | histogram,
records | Number of rows fetched per successful poll, including empty
polls.
|
+| `batch-size-records` | histogram,
records | Number of rows fetched per successful poll, including empty
polls. Polls at `pollBatchSize` mean the server is draining a backlog and polls
again at once. |
| `poll-duration` | timer,
duration | End-to-end poll-cycle duration.
|
| `invalidation-failures-total`, `fallback-clears-total` | counter,
failures/clears | Failed targeted entity-cache invalidations and successful
full-cache recovery clears.
|
@@ -102,9 +103,11 @@ recover locally; its failure counter and log identify the
affected listener. The
row was added to the current transaction, not that the transaction committed.
For an incident, check `seconds-since-last-successful-tail-sample` before
comparing `db-tail-id`
-with `cursor-id` on the affected server. If it exceeds the poll interval, the
tail sample itself is
-failing: the retained tail can fall below an advancing cursor and `record-lag`
can read zero despite
-an unknown database tail. `tail-sample-failures-total` counts those failures
for alerting. With a
+with `cursor-id` on the affected server. If it substantially exceeds the
normal sampling interval,
+investigate slow or failed polling and tail sampling;
`tail-sample-failures-total` counts actual
+sample failures. Between samples, an advancing cursor can pass the retained
tail, so `record-lag`
+can read zero even while full batches remain. Use `batch-size-records` at
`pollBatchSize` as the
+backlog signal rather than treating zero sampled lag as proof that the poller
has caught up. With a
fresh tail sample, a growing `record-lag` together with an increasing poll age
or
`poll-failures-total` points to polling trouble.
If the cursor advances but data remains stale, inspect
`listener-failures-total`, `records-delivered.<class>-total`,
