jerryshao commented on code in PR #13619:
URL: https://github.com/apache/gravitino/pull/13619#discussion_r4215479315
##########
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:
[Nit] This comment overstates what the policy does, and the overstatement is
the kind a future reader could act on.
`ScheduledThreadPoolExecutor.onShutdown()` only removes a non-periodic task
when `!keepDelayed && t.getDelay(NANOSECONDS) > 0`. A task whose delay has
already elapsed is kept and still runs after `shutdown()`, regardless of this
policy. Every drain hop schedules with delay `0` (`runPollCycle`, line 238), so
for a poller that is actively draining a backlog -- the exact case this PR adds
-- `close()` does leave an already-due task that runs.
It is harmless today, but only because `runPollCycle` returns at line 232 on
the `closed` flag, and `close()` sets `closed` before `shutdown()`. So the real
guarantee is the `closed` guard; the policy only helps the idle case. Worth
saying that here, so nobody later reads the `closed` check as redundant with
the policy and deletes it. `testCloseDropsThePendingPollInsteadOfWaitingForIt`
only covers the idle case too (interval 2 s, nothing due), so it would not
catch that regression.
Verified by: compiled a standalone probe against the installed JDK (21.0.12)
-- with `setExecuteExistingDelayedTasksAfterShutdownPolicy(false)`, a task
queued at delay 0 behind a busy worker still ran after `shutdown()`
(`taskRanAfterShutdown=true`), while a task 2 s out was dropped and
`shutdown()` returned in 0 ms. The `getDelay() > 0` condition in `onShutdown()`
is unchanged between JDK 17 and 21.
##########
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:
[Nit] This paragraph is good, but it leaves the in-repo design doc
contradicting it.
`design-docs/gravitino-entity-cache-multinode-design.md:341` still describes
the old fixed-rate model: "every other node sees it after the next successful
poll. The interval is set by `gravitino.entityChangeLog.pollIntervalSecs`" --
no mention of `pollBatchSize` or of backlog draining, which is precisely what
this PR changes. One sentence there, pointing at the behaviour you just
documented here, would keep the two in sync.
Verified by: read
`design-docs/gravitino-entity-cache-multinode-design.md:336-343`; the "Expected
propagation and failure behavior" section is untouched by this PR (`git diff
origin/main...HEAD` lists no `design-docs/` file).
##########
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:
[Question] Was DB pressure during a sustained drain considered? There is no
floor on the delay and no cap on consecutive immediate hops, so a server whose
write rate stays at or above `pollBatchSize` per cycle polls as fast as the
database answers, indefinitely.
Each cycle issues two statements, not one: `selectEntityChanges` plus the
best-effort tail sample `selectMaxChangeId` at lines 304-307, each through its
own `SessionUtils.getWithoutCommit` session. The tail sample is
observability-only (your own comment at 300-301 says so) and during a drain it
tells you nothing new, since a full batch is already the backlog signal that
line 293 acts on. Skipping it while the previous cycle reported a backlog, or
giving the zero-delay hop a small floor (a few ms), would cut the steady-state
query rate roughly in half for no loss of drain speed.
Not blocking -- continuous draining is the point of the PR, and the cursor
always advances so this cannot become a spin. I am asking whether the unbounded
rate is deliberate.
Verified by: read `runPollCycle` (231-245), `doPollChanges` (284-294) and
`fetchNextDelivery` (297-341) in full; confirmed both statements run on every
cycle regardless of the delay chosen, and confirmed via
`EntityChangeLogBaseSQLProvider.java:58,62` that they are two separate queries.
--
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]
