yuqi1129 commented on code in PR #13619:
URL: https://github.com/apache/gravitino/pull/13619#discussion_r4217443538
##########
core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java:
##########
@@ -190,28 +207,90 @@ public void close() {
}
@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;
+ });
+ // Each poll is a one-shot delayed task. By default such tasks still run
after shutdown(), so
+ // close() would wait for the pending poll and run it against a store that
is shutting down.
+ executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
Review Comment:
Thanks, good catch. Fixed in ec682a31dd. The comment on `createScheduler`
now says that the policy only drops polls whose delay has not elapsed.
Already-due tasks, including zero-delay drain hops, can still run after
`shutdown()`, and it is the `closed` guard in `runPollCycle` that keeps them
away from the closing store. I also added
`testAlreadyDueDrainHopDoesNotPollAfterClose`. It uses the production
scheduler, queues an already-due drain hop behind a busy worker during
`close()`, and asserts that the hop executes without polling. Removing the
`closed` guard makes that test fail.
##########
core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java:
##########
@@ -190,28 +207,90 @@ public void close() {
}
@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;
+ });
+ // Each poll is a one-shot delayed task. By default such tasks still run
after shutdown(), so
+ // close() would wait for the pending poll and run it against a store that
is shutting down.
+ executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+ return executor;
+ }
+
+ /**
+ * 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()) {
+ nextDelayMillis = 0;
+ }
Review Comment:
Yes, continuous draining is deliberate: every cycle is bounded by
`pollBatchSize` and always advances the cursor, so a sustained backlog is
drained at the rate the database and listeners support. That is now stated
explicitly in `docs/gravitino-server-config.md` (Change Log Propagation) and in
the PR description. I took your suggestion on the tail query (ec682a31dd):
during a full-batch drain, `selectMaxChangeId` is sampled at most once per
`pollIntervalSecs`, and failed attempts count toward that limit too. Partial
and empty polls still refresh the tail every cycle, so a sustained drain issues
roughly one query per cycle instead of two. This is covered by
`testDrainSamplesTailOncePerIntervalAndRefreshesAfterCatchingUp` and
`testFailedTailSamplesAreAlsoRateLimitedDuringDrain`. `docs/metrics.md`
(32184fed76) now explains that `record-lag` is sampled, and that
`batch-size-records` at `pollBatchSize` is the backlog signal. I did not add a
delay floor, to keep drain throughput indepen
dent of an arbitrary constant.
##########
docs/gravitino-server-config.md:
##########
@@ -326,9 +326,18 @@ 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. Each poll holds its whole batch in memory, so raise `pollBatchSize` in
moderate steps.
Review Comment:
Done in ec682a31dd. The "Expected propagation and failure behavior" bullet
in `design-docs/gravitino-entity-cache-multinode-design.md` now describes
`pollBatchSize`, the immediate next poll after a full batch, and the interval
after an empty, partial, or failed poll. It also links to the Change Log
Propagation section of the server config doc for the capacity and database-load
trade-offs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
