Copilot commented on code in PR #2598:
URL: https://github.com/apache/phoenix/pull/2598#discussion_r3786988851
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java:
##########
@@ -196,6 +226,61 @@ public void stop() {
LOG.info("ReplicationLogDiscovery stopped for haGroup: {}", haGroupName);
}
+ /**
+ * Schedules the next replay as a single-shot task whose delay is recomputed
each cycle via
+ * {@link #computeAlignedInitialDelay()}. Recomputing every cycle re-pins
each wake-up to the
+ * wall-clock round-eligibility grid, correcting scheduler/wall-clock drift
instead of letting a
+ * one-time misalignment persist for the life of the process (which
fixed-rate scheduling does).
+ * All region servers still converge on the same grid, preserving
PHOENIX-7813's shared wake-up.
+ */
+ @GuardedBy("this")
+ protected void scheduleNextReplay() {
+ long delayMs = computeAlignedInitialDelay();
+ // Bind this cycle to the current scheduler generation. A stop()->start()
restart
+ // swaps in a new scheduler; a cycle launched on the old one must
reschedule onto
+ // that same (now shut-down) scheduler, not the new one.
+ ScheduledExecutorService owner = scheduler;
+ LOG.info("Scheduling next replay for haGroup: {} in {}ms", haGroupName,
delayMs);
+ owner.schedule(() -> runReplayCycle(owner), delayMs,
TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Runs one replay pass and, unless the service has been stopped, schedules
the next aligned pass.
+ * Exceptions from {@link #replay()} are swallowed so a single failure does
not break the chain.
+ * The reschedule is guarded by the same lock stop() uses; if stop() shut
the scheduler down
+ * first, {@link #isRunning} is false and we do not reschedule (and a
concurrent shutdown that
+ * rejects the submission is caught and treated as "stop the chain").
+ * @param owner the scheduler this cycle was launched on. If a
stop()->start() restart has since
+ * swapped in a new scheduler, {@code owner} no longer equals
{@link #scheduler} and
+ * this stale cycle must not reschedule onto the new generation
(which would create a
+ * second concurrent chain and double the effective poll rate).
+ */
+ protected void runReplayCycle(ScheduledExecutorService owner) {
+ try {
+ replay();
+ } catch (Throwable t) {
+ LOG.error("Error during replay for haGroup: {}", haGroupName, t);
+ } finally {
+ synchronized (this) {
+ if (isRunning && owner == scheduler) {
+ try {
+ scheduleNextReplay();
+ } catch (RejectedExecutionException ree) {
+ // benign: stop() shut the scheduler down between the guard check
and submit
+ LOG.debug("Scheduler shutting down, skipping reschedule for
haGroup: {}", haGroupName);
+ } catch (Throwable t) {
+ // Any other failure (e.g. a bad epsilon config value making
+ // computeAlignedInitialDelay throw) would otherwise be swallowed
by the executor
+ // into the discarded Future and silently wedge the polling chain
with
+ // isRunning==true -- the exact silent-stop this class is meant to
prevent.
Review Comment:
This failure terminates the one-shot chain but leaves `isRunning == true`
and a live idle executor. A later `start()` therefore returns as “already
running,” so logging the exception still leaves polling permanently wedged.
Either retry with a bounded fallback/backoff or atomically mark the service
stopped and shut down the owner so it can be restarted.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java:
##########
@@ -544,6 +624,16 @@ public int getInProgressFileMinAgeSeconds() {
DEFAULT_IN_PROGRESS_FILE_MIN_AGE_SECONDS);
}
+ /**
+ * Returns the epsilon margin (milliseconds) added to the aligned scheduler
wake instant.
+ * @return the epsilon margin in milliseconds (default
+ * {@link #DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS}).
+ */
+ public long getAlignedDelayEpsilonMillis() {
+ return conf.getLong(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY,
+ DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS);
+ }
Review Comment:
The new config is used as an offset without validating its range. A negative
value moves wakes before eligibility, while a value greater than or equal to
the round duration wraps through `floorMod` and no longer represents the
documented epsilon after the boundary. Reject values outside `[0,
roundTimeMills)` so a misconfiguration cannot silently reintroduce missed
rounds.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java:
##########
@@ -196,6 +226,61 @@ public void stop() {
LOG.info("ReplicationLogDiscovery stopped for haGroup: {}", haGroupName);
}
+ /**
+ * Schedules the next replay as a single-shot task whose delay is recomputed
each cycle via
+ * {@link #computeAlignedInitialDelay()}. Recomputing every cycle re-pins
each wake-up to the
+ * wall-clock round-eligibility grid, correcting scheduler/wall-clock drift
instead of letting a
+ * one-time misalignment persist for the life of the process (which
fixed-rate scheduling does).
+ * All region servers still converge on the same grid, preserving
PHOENIX-7813's shared wake-up.
+ */
+ @GuardedBy("this")
+ protected void scheduleNextReplay() {
+ long delayMs = computeAlignedInitialDelay();
Review Comment:
After a cycle completes, this can schedule the same grid point again.
`computeAlignedInitialDelay()` returns `0` at the exact tick and a small
positive delay when the nanoTime-based wake arrives slightly before the
epsilon-shifted wall-clock tick, so a fast replay can run repeatedly for one
round (including duplicate failover/state checks) instead of advancing to the
next round. Carry the intended wall-clock target into the callback and make
post-cycle scheduling select a boundary strictly after that cycle's target.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java:
##########
@@ -196,6 +226,61 @@ public void stop() {
LOG.info("ReplicationLogDiscovery stopped for haGroup: {}", haGroupName);
}
+ /**
+ * Schedules the next replay as a single-shot task whose delay is recomputed
each cycle via
+ * {@link #computeAlignedInitialDelay()}. Recomputing every cycle re-pins
each wake-up to the
+ * wall-clock round-eligibility grid, correcting scheduler/wall-clock drift
instead of letting a
+ * one-time misalignment persist for the life of the process (which
fixed-rate scheduling does).
+ * All region servers still converge on the same grid, preserving
PHOENIX-7813's shared wake-up.
+ */
+ @GuardedBy("this")
+ protected void scheduleNextReplay() {
+ long delayMs = computeAlignedInitialDelay();
+ // Bind this cycle to the current scheduler generation. A stop()->start()
restart
+ // swaps in a new scheduler; a cycle launched on the old one must
reschedule onto
+ // that same (now shut-down) scheduler, not the new one.
+ ScheduledExecutorService owner = scheduler;
+ LOG.info("Scheduling next replay for haGroup: {} in {}ms", haGroupName,
delayMs);
+ owner.schedule(() -> runReplayCycle(owner), delayMs,
TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Runs one replay pass and, unless the service has been stopped, schedules
the next aligned pass.
+ * Exceptions from {@link #replay()} are swallowed so a single failure does
not break the chain.
+ * The reschedule is guarded by the same lock stop() uses; if stop() shut
the scheduler down
+ * first, {@link #isRunning} is false and we do not reschedule (and a
concurrent shutdown that
+ * rejects the submission is caught and treated as "stop the chain").
+ * @param owner the scheduler this cycle was launched on. If a
stop()->start() restart has since
+ * swapped in a new scheduler, {@code owner} no longer equals
{@link #scheduler} and
+ * this stale cycle must not reschedule onto the new generation
(which would create a
+ * second concurrent chain and double the effective poll rate).
+ */
+ protected void runReplayCycle(ScheduledExecutorService owner) {
+ try {
+ replay();
+ } catch (Throwable t) {
+ LOG.error("Error during replay for haGroup: {}", haGroupName, t);
Review Comment:
Catching `Throwable` makes fatal JVM conditions such as `OutOfMemoryError`,
`StackOverflowError`, and linkage failures look recoverable, and the `finally`
block then schedules another replay on the potentially corrupted JVM. Catch
expected `Exception` failures for chain continuity, but do not swallow fatal
`Error`s or reschedule after them.
--
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]
