apurtell commented on code in PR #2598:
URL: https://github.com/apache/phoenix/pull/2598#discussion_r3787116680
##########
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);
Review Comment:
No.
Audit all your code for this kind of noisy and low value INFO level logging.
It must be DEBUG, if not TRACE.
##########
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) {
Review Comment:
`isRunning` remains true and the executor stays alive, so
`discovery.isRunning()` reports healthy while nothing happens.
At least set `isRunning = false` in the catch handler.
Another option is to log a warning at WARN log level and fall back to
`DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS` on `NumberFormatException` (or anything
that is not `RejectedExecutionException`) and retry. This rides over
configuration errors.
##########
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:
Agreed
--
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]
