From 72951b0537f6d1a63631b2ec7995968d859a1bb6 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 13 Aug 2026 01:19:41 +0530 Subject: [PATCH] PHOENIX-7978 Harden replay/forward poll scheduling against wall-clock and scheduler drift The round-eligibility gate for replication replay/forward becomes eligible when currentTime - lastRoundEndTimestamp >= roundTimeMills + bufferMillis, evaluated on the wall clock. PHOENIX-7813 aligned the scheduler wake to that grid, but the wake is fired on the monotonic clock (System.nanoTime) and was computed with zero margin, so small nanoTime-vs-wall-clock drift could tip a wake just below the boundary and the region server would lose a full (~60s) cycle. Most damaging during planned failover. Two fixes, both in the shared base class ReplicationLogDiscovery (inherited by ReplicationLogDiscoveryReplay and ReplicationLogDiscoveryForwarder): - Epsilon margin on the aligned wake instant: anchor the delay at bufferMillis + epsilon (via Math.floorMod) so the wake lands just after the eligibility boundary rather than exactly on it. New config phoenix.replication.discovery.aligned.delay.epsilon.millis (default 500). - Per-cycle re-anchor: replace scheduleAtFixedRate with a self-rescheduling one-shot chain that recomputes the aligned delay every cycle, re-pinning each wake to the wall-clock grid instead of letting a one-time misalignment persist. Uses a ScheduledThreadPoolExecutor with setExecuteExistingDelayedTasksAfterShutdownPolicy(false) so stop() is deterministic. Each replay cycle is bound to the scheduler generation it was launched on and reschedules only if isRunning && owner == scheduler, preventing a stale in-flight cycle from grafting a second chain onto a new scheduler after a stop()->start() restart (which would otherwise double the effective poll rate). Testing: ReplicationLogDiscoveryTest 48/48 (incl. stale-generation, start()-rollback, and replay/reschedule error-swallow regressions); ReplicationLogDiscoveryReplayTestIT 48/48; StoreAndForwardFailoverIT 1/1; spotless:check green on phoenix-core and phoenix-core-server. --- .../replication/ReplicationLogDiscovery.java | 152 ++++++++-- .../ReplicationLogDiscoveryReplayTestIT.java | 29 -- .../ReplicationLogDiscoveryTest.java | 268 ++++++++++++++++-- 3 files changed, 367 insertions(+), 82 deletions(-) diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java index 638d95280f5..53af2a7e26b 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java @@ -22,10 +22,12 @@ import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import javax.annotation.concurrent.GuardedBy; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; @@ -99,9 +101,32 @@ public abstract class ReplicationLogDiscovery { public static final int DEFAULT_IN_PROGRESS_FILE_MIN_AGE_SECONDS = 60; + /** + * Configuration key for the epsilon margin (milliseconds) added to the aligned scheduler wake + * instant. The replay scheduler fires on a {@code System.nanoTime()} grid while the round + * eligibility gate reads the wall clock ({@code EnvironmentEdgeManager.currentTime()}). Aligning + * exactly to the eligibility instant lets a few ms of nanoTime-vs-wall-clock skew tip a wake-up + * just below the boundary, which costs a full poll cycle. Waking epsilon after the eligibility + * instant absorbs that skew. + */ + public static final String REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY = + "phoenix.replication.discovery.aligned.delay.epsilon.millis"; + + /** + * Default epsilon margin in milliseconds. 500ms comfortably exceeds the small (single- to + * low-tens-of-milliseconds) nanoTime-vs-wall-clock skew this margin absorbs, yet stays under 1% + * of a 60s round, so best-case first-pickup latency is essentially unchanged. The margin is + * absolute (it offsets clock skew, which does not scale with round duration), so for atypically + * short custom round durations operators may lower it via + * {@link #REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY} to keep epsilon a small fraction of the + * round. + */ + public static final long DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS = 500L; + protected final Configuration conf; protected final String haGroupName; protected final ReplicationLogTracker replicationLogTracker; + @GuardedBy("this") protected ScheduledExecutorService scheduler; protected volatile boolean isRunning = false; protected volatile ReplicationRound lastRoundProcessed; @@ -132,9 +157,10 @@ public void close() { } /** - * Starts the replication log discovery service by initializing the scheduler and scheduling - * periodic replay operations. Creates a thread pool with configured thread count and schedules - * replay tasks at fixed intervals. + * Starts the replication log discovery service. Creates a scheduler with the configured thread + * count and launches a self-rescheduling one-shot replay chain (see + * {@link #scheduleNextReplay()}) that re-anchors each replay to the aligned round-eligibility + * grid every cycle, rather than firing at a fixed period. * @throws IOException if there's an error during initialization */ public void start() throws IOException { @@ -143,22 +169,26 @@ public void start() throws IOException { LOG.warn("ReplicationLogDiscovery is already running for haGroup: {}", haGroupName); return; } - // Initialize and schedule the executors - scheduler = Executors.newScheduledThreadPool(getExecutorThreadCount(), - new ThreadFactoryBuilder().setNameFormat(getExecutorThreadNameFormat()).build()); - long initialDelayMs = computeAlignedInitialDelay(); - long replayIntervalMs = getReplayIntervalMillis(); - LOG.info("Scheduling replay for haGroup: {} with initialDelay={}ms, interval={}ms", - haGroupName, initialDelayMs, replayIntervalMs); - scheduler.scheduleAtFixedRate(() -> { - try { - replay(); - } catch (Exception e) { - LOG.error("Error during replay", e); - } - }, initialDelayMs, replayIntervalMs, TimeUnit.MILLISECONDS); - + // Single-shot rescheduling chain (see scheduleNextReplay). Discard any queued + // (not-yet-started) delayed task on shutdown so stop() is deterministic and no replay + // fires after we intend to stop. + ScheduledThreadPoolExecutor executor = + new ScheduledThreadPoolExecutor(getExecutorThreadCount(), + new ThreadFactoryBuilder().setNameFormat(getExecutorThreadNameFormat()).build()); + executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + scheduler = executor; isRunning = true; + try { + scheduleNextReplay(); + } catch (RuntimeException | Error e) { + // Scheduling the first cycle failed (e.g. a bad epsilon config value). Roll back so we + // don't leave a live idle executor with isRunning==true (which reports healthy while + // nothing polls) and so a later start() can retry cleanly. + isRunning = false; + scheduler = null; + executor.shutdownNow(); + throw e; + } LOG.info("ReplicationLogDiscovery started for haGroup: {}", haGroupName); } } @@ -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. + LOG.error("Failed to schedule next replay for haGroup: {}; replay polling has stopped", + haGroupName, t); + } + } + } + } + } + /** * Executes a replay operation for the next set of replication rounds. This method continuously * retrieves and processes rounds using getNextRoundToProcess() until: - No more rounds are ready @@ -484,15 +569,6 @@ public String getExecutorThreadNameFormat() { return DEFAULT_EXECUTOR_THREAD_NAME_FORMAT; } - /** - * Returns the replay interval in milliseconds. Subclasses can override this method to provide - * custom intervals. Defaults to the round duration. - * @return The replay interval in milliseconds. - */ - public long getReplayIntervalMillis() { - return roundTimeMills; - } - /** * Returns the shutdown timeout in seconds. Subclasses can override this method to provide custom * timeout values. @@ -524,13 +600,17 @@ public double getWaitingBufferPercentage() { * Computes initial delay to align the scheduler to round-eligible boundaries so all RS wake up at * the same wall-clock moment. A round becomes eligible when currentTime >= roundEndTime + * bufferMillis, and rounds repeat every roundTimeMills. This gives a universal grid of eligible - * ticks at bufferMillis, bufferMillis + roundTimeMills, bufferMillis + 2*roundTimeMills, etc. - * from epoch. All RS compute the same grid regardless of when start() is called. + * ticks at bufferMillis + epsilon, bufferMillis + epsilon + roundTimeMills, bufferMillis + + * epsilon + 2*roundTimeMills, etc. from epoch. All RS compute the same grid regardless of when + * start() is called. * @return the initial delay in milliseconds until the next round-eligible tick */ protected long computeAlignedInitialDelay() { long now = EnvironmentEdgeManager.currentTime(); - long elapsed = (now - bufferMillis) % roundTimeMills; + // Anchor epsilon past the eligibility instant (bufferMillis past a round line) so that a + // scheduler firing slightly early (nanoTime skew) still clears the wall-clock gate. + long anchor = bufferMillis + getAlignedDelayEpsilonMillis(); + long elapsed = Math.floorMod(now - anchor, roundTimeMills); return (elapsed == 0) ? 0 : roundTimeMills - elapsed; } @@ -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); + } + public ReplicationLogTracker getReplicationLogFileTracker() { return this.replicationLogTracker; } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java index 98209280f48..e65525d284b 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java @@ -116,35 +116,6 @@ public void testGetExecutorThreadNameFormat() throws IOException { "Phoenix-ReplicationLogDiscoveryReplay-%d", result); } - /** - * Tests that replay interval always matches the configured round duration. - */ - @Test - public void testGetReplayIntervalMillis() throws IOException { - // Test with default round duration - TestableReplicationLogTracker fileTracker = - createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri); - ReplicationLogDiscoveryReplay discovery = new ReplicationLogDiscoveryReplay(fileTracker); - long expectedRoundMillis = - fileTracker.getReplicationShardDirectoryManager().getReplicationRoundDurationSeconds() - * 1000L; - assertEquals("Replay interval should match round duration", expectedRoundMillis, - discovery.getReplayIntervalMillis()); - - // Test with custom round duration - conf1.setInt(ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY, - 120); - try { - TestableReplicationLogTracker fileTracker2 = - createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri); - ReplicationLogDiscoveryReplay discovery2 = new ReplicationLogDiscoveryReplay(fileTracker2); - assertEquals("Replay interval should match custom round duration", 120_000L, - discovery2.getReplayIntervalMillis()); - } finally { - conf1.unset(ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY); - } - } - /** * Tests the shutdown timeout configuration with default and custom values. */ diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java index 4af410e56e3..5851d360fff 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java @@ -139,10 +139,6 @@ public void testStartAndStop() throws IOException { assertTrue("Thread name should contain ReplicationLogDiscovery", threadName.contains("ReplicationLogDiscovery")); - // Verify replay interval - long replayInterval = discovery.getReplayIntervalMillis(); - assertEquals("Replay interval should be 60000 milliseconds", 60_000L, replayInterval); - // 6. Ensure starting again does not create a new scheduler (and also should not throw any // exception) ScheduledExecutorService originalScheduler = discovery.getScheduler(); @@ -161,15 +157,159 @@ public void testStartAndStop() throws IOException { assertFalse("Discovery should not be running after stop", discovery.isRunning()); } + @Test + public void testStartRollsBackWhenSchedulingFails() throws IOException { + // setUp() stubs isRunning() to always return true; read the real field for this lifecycle test. + Mockito.doCallRealMethod().when(discovery).isRunning(); + doThrow(new NumberFormatException("bad epsilon")).when(discovery).scheduleNextReplay(); + NumberFormatException thrown = null; + try { + discovery.start(); + } catch (NumberFormatException e) { + thrown = e; + } + assertNotNull("start() should propagate the scheduling failure", thrown); + assertFalse("start() must roll back isRunning on failure", discovery.isRunning()); + assertNull("start() must roll back the scheduler on failure", discovery.getScheduler()); + } + + @Test + public void testRunReplayCycleReschedulesWhenRunning() throws IOException { + ScheduledExecutorService owner = mock(ScheduledExecutorService.class); + discovery.setScheduler(owner); + discovery.setRunning(true); + doNothing().when(discovery).replay(); + doNothing().when(discovery).scheduleNextReplay(); + + discovery.runReplayCycle(owner); + + verify(discovery, times(1)).replay(); + verify(discovery, times(1)).scheduleNextReplay(); + } + + @Test + public void testRunReplayCycleDoesNotRescheduleWhenStopped() throws IOException { + ScheduledExecutorService owner = mock(ScheduledExecutorService.class); + discovery.setScheduler(owner); + discovery.setRunning(false); + doNothing().when(discovery).replay(); + doNothing().when(discovery).scheduleNextReplay(); + + discovery.runReplayCycle(owner); + + verify(discovery, times(1)).replay(); + verify(discovery, never()).scheduleNextReplay(); + } + + @Test + public void testRunReplayCycleReschedulesAfterReplayThrows() throws IOException { + ScheduledExecutorService owner = mock(ScheduledExecutorService.class); + discovery.setScheduler(owner); + discovery.setRunning(true); + doThrow(new IOException("boom")).when(discovery).replay(); + doNothing().when(discovery).scheduleNextReplay(); + + discovery.runReplayCycle(owner); // must not propagate the exception + + verify(discovery, times(1)).replay(); + verify(discovery, times(1)).scheduleNextReplay(); + } + + @Test + public void testRunReplayCycleReschedulesAfterReplayThrowsError() throws IOException { + ScheduledExecutorService owner = mock(ScheduledExecutorService.class); + discovery.setScheduler(owner); + discovery.setRunning(true); + // An Error (OOME/StackOverflow/linkage) from replay() must be caught and logged, not slip + // past catch (Exception) and vanish into the executor's discarded Future. + doThrow(new Error("boom")).when(discovery).replay(); + doNothing().when(discovery).scheduleNextReplay(); + + discovery.runReplayCycle(owner); // must not propagate the Error + + verify(discovery, times(1)).replay(); + verify(discovery, times(1)).scheduleNextReplay(); + } + + @Test + public void testRunReplayCycleDoesNotPropagateWhenRescheduleThrows() throws IOException { + ScheduledExecutorService owner = mock(ScheduledExecutorService.class); + discovery.setScheduler(owner); + discovery.setRunning(true); + doNothing().when(discovery).replay(); + // A bad epsilon config value makes the reschedule path (computeAlignedInitialDelay -> + // getLong) throw NumberFormatException. It must be caught, not swallowed by the executor + // into the discarded Future, which would silently wedge the chain with isRunning==true. + doThrow(new NumberFormatException("bad epsilon")).when(discovery).scheduleNextReplay(); + + discovery.runReplayCycle(owner); // must not propagate the reschedule failure + + verify(discovery, times(1)).replay(); + verify(discovery, times(1)).scheduleNextReplay(); + } + + @Test + public void testRunReplayCycleDoesNotRescheduleForStaleSchedulerGeneration() throws IOException { + ScheduledExecutorService staleOwner = mock(ScheduledExecutorService.class); + ScheduledExecutorService currentScheduler = mock(ScheduledExecutorService.class); + discovery.setScheduler(currentScheduler); + discovery.setRunning(true); + doNothing().when(discovery).replay(); + doNothing().when(discovery).scheduleNextReplay(); + + discovery.runReplayCycle(staleOwner); // stale generation: owner != current scheduler + + verify(discovery, times(1)).replay(); + verify(discovery, never()).scheduleNextReplay(); + } + + @Test + public void testStaleCycleDoesNotRescheduleAfterRealRestart() throws IOException { + // Real start()/stop()/start() establishes the generation state (not hand-set fields). + // Far-future aligned delay => the auto-scheduled cycle never fires during the test. + doReturn(TimeUnit.HOURS.toMillis(1)).when(discovery).computeAlignedInitialDelay(); + doNothing().when(discovery).replay(); + + discovery.start(); // gen-1 + ScheduledExecutorService s1 = discovery.getScheduler(); + discovery.stop(); // shuts down s1 + discovery.start(); // gen-2, isRunning flipped back true + ScheduledExecutorService s2 = discovery.getScheduler(); + assertTrue("restart must create a new scheduler generation", s1 != s2); + assertTrue("old scheduler must be shut down", s1.isShutdown()); + assertTrue("discovery must be running after restart", discovery.isRunning()); + + // gen-1's in-flight cycle reaches its finally AFTER the restart: isRunning is true again, + // but owner(s1) != scheduler(s2), so it must NOT graft a second chain onto s2. + discovery.runReplayCycle(s1); + + verify(discovery, times(2)).scheduleNextReplay(); // one per start(); stale cycle adds none + discovery.stop(); + } + + @Test + public void testScheduleNextReplayUsesAlignedDelay() { + ScheduledExecutorService mockScheduler = mock(ScheduledExecutorService.class); + discovery.setScheduler(mockScheduler); + long knownDelay = 1_234L; + doReturn(knownDelay).when(discovery).computeAlignedInitialDelay(); + + discovery.scheduleNextReplay(); + + verify(mockScheduler, times(1)).schedule(any(Runnable.class), eq(knownDelay), + eq(TimeUnit.MILLISECONDS)); + } + @Test public void testComputeAlignedInitialDelay() { long roundTimeMs = discovery.roundTimeMills; long bufferMs = discovery.bufferMillis; + long epsilon = discovery.getAlignedDelayEpsilonMillis(); // RS initialize at different times within the same round window. // All should align to the same next tick. - // With roundTimeMs=60000 and bufferMs=9000, ticks are at 9000, 69000, 129000, ... - // Place all 3 RS between tick 69000 and tick 129000 so they all target 129000. + // With roundTimeMs=60000 and bufferMs=9000, ticks are at 9000+epsilon, 69000+epsilon, ... + // Place all 3 RS between tick 69000+epsilon and 129000+epsilon so they target 129000+epsilon. AtomicLong mockTime = new AtomicLong(); EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() { @Override @@ -194,10 +334,10 @@ public long currentTime() { long delay3 = discovery.computeAlignedInitialDelay(); long tick3 = mockTime.get() + delay3; - // All should align to the same tick (129000) + // All should align to the same tick (129000 + epsilon) assertEquals("RS-1 and RS-2 should align to the same tick", tick1, tick2); assertEquals("RS-2 and RS-3 should align to the same tick", tick2, tick3); - assertEquals("All should target tick at 129000", 129_000L, tick1); + assertEquals("All should target the epsilon-shifted tick", 129_000L + epsilon, tick1); // Delay should always be > 0 and <= roundTimeMs assertTrue("Delay should be positive", delay1 > 0); @@ -205,9 +345,37 @@ public long currentTime() { assertTrue("Delay should be positive", delay2 > 0); assertTrue("Delay should not exceed round time", delay2 <= roundTimeMs); - // The aligned tick should be at a multiple of roundTimeMs offset by bufferMs - assertEquals("Tick should be aligned to round-eligible boundary", 0, - (tick1 - bufferMs) % roundTimeMs); + // The aligned tick should be at a multiple of roundTimeMs offset by bufferMs + epsilon + assertEquals("Tick should be aligned to the epsilon-shifted grid", 0, + (tick1 - bufferMs - epsilon) % roundTimeMs); + } finally { + EnvironmentEdgeManager.reset(); + } + } + + @Test + public void testComputeAlignedInitialDelayWhenNowBeforeAnchor() { + // Regression lock-in for Math.floorMod: when now < the first anchor (bufferMillis + epsilon), + // now - anchor is negative; floorMod keeps elapsed in [0, roundTimeMills) so we target the + // first tick. Plain (now - bufferMillis) % roundTimeMills would go negative -> delay > a round. + long roundTimeMs = discovery.roundTimeMills; + long bufferMs = discovery.bufferMillis; + long epsilon = discovery.getAlignedDelayEpsilonMillis(); + AtomicLong mockTime = new AtomicLong(); + EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() { + @Override + public long currentTime() { + return mockTime.get(); + } + }); + try { + mockTime.set(100L); // before the very first anchor (bufferMs + epsilon = 9500) + long delay = discovery.computeAlignedInitialDelay(); + long targetTick = 100L + delay; + assertTrue("Delay should be positive", delay > 0); + assertTrue("Delay should not exceed round time", delay <= roundTimeMs); + assertEquals("Should target the first epsilon-shifted tick", bufferMs + epsilon, targetTick); + assertEquals("Target should be aligned", 0, (targetTick - bufferMs - epsilon) % roundTimeMs); } finally { EnvironmentEdgeManager.reset(); } @@ -227,11 +395,12 @@ public long currentTime() { }); try { - // Set time to exactly on a round-eligible tick boundary - long exactTick = roundTimeMs * 5 + bufferMs; + // Set time to exactly on an epsilon-shifted round-eligible tick boundary + long epsilon = discovery.getAlignedDelayEpsilonMillis(); + long exactTick = roundTimeMs * 5 + bufferMs + epsilon; mockTime.set(exactTick); long delay = discovery.computeAlignedInitialDelay(); - assertEquals("Delay should be 0 when exactly on a tick", 0, delay); + assertEquals("Delay should be 0 when exactly on an epsilon-shifted tick", 0, delay); } finally { EnvironmentEdgeManager.reset(); } @@ -253,21 +422,22 @@ public long currentTime() { }); try { - // Round-eligible tick is at roundTimeMs * 5 + bufferMs + // Round-eligible tick is at roundTimeMs * 5 + bufferMs + epsilon // Set time to 3s past that tick (12s into the round, buffer is 9s) - long tick = roundTimeMs * 5 + bufferMs; + long epsilon = discovery.getAlignedDelayEpsilonMillis(); + long tick = roundTimeMs * 5 + bufferMs + epsilon; long now = tick + 3_000L; mockTime.set(now); long delay = discovery.computeAlignedInitialDelay(); // Should wait until the next tick: tick + roundTimeMs long expectedDelay = roundTimeMs - 3_000L; - assertEquals("Should wait until next round-eligible tick", expectedDelay, delay); + assertEquals("Should wait until next epsilon-shifted tick", expectedDelay, delay); // Verify the target tick is correct long targetTick = now + delay; assertEquals("Target should be next tick", tick + roundTimeMs, targetTick); - assertEquals("Target should be aligned", 0, (targetTick - bufferMs) % roundTimeMs); + assertEquals("Target should be aligned", 0, (targetTick - bufferMs - epsilon) % roundTimeMs); } finally { EnvironmentEdgeManager.reset(); } @@ -289,20 +459,66 @@ public long currentTime() { }); try { - // Round-eligible tick is at roundTimeMs * 5 + bufferMs + // Round-eligible tick is at roundTimeMs * 5 + bufferMs + epsilon // Set time to 3s before that tick (6s into the round, buffer is 9s) - long tick = roundTimeMs * 5 + bufferMs; + long epsilon = discovery.getAlignedDelayEpsilonMillis(); + long tick = roundTimeMs * 5 + bufferMs + epsilon; long now = tick - 3_000L; mockTime.set(now); long delay = discovery.computeAlignedInitialDelay(); // Should wait 3s until the upcoming tick - assertEquals("Should wait until upcoming round-eligible tick", 3_000L, delay); + assertEquals("Should wait until upcoming epsilon-shifted tick", 3_000L, delay); // Verify the target tick is correct long targetTick = now + delay; assertEquals("Target should be the upcoming tick", tick, targetTick); - assertEquals("Target should be aligned", 0, (targetTick - bufferMs) % roundTimeMs); + assertEquals("Target should be aligned", 0, (targetTick - bufferMs - epsilon) % roundTimeMs); + } finally { + EnvironmentEdgeManager.reset(); + } + } + + @Test + public void testAlignedDelayEpsilonDefaultAndConfig() { + // Default value. + assertEquals("Default epsilon should be 500ms", + ReplicationLogDiscovery.DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS, + discovery.getAlignedDelayEpsilonMillis()); + + // Custom value from configuration. + conf.setLong(ReplicationLogDiscovery.REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY, 1_500L); + assertEquals("Epsilon should be read from config", 1_500L, + discovery.getAlignedDelayEpsilonMillis()); + conf.unset(ReplicationLogDiscovery.REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY); + } + + @Test + public void testComputeAlignedInitialDelayWakesEpsilonAfterEligibility() { + // Reproduces the near-miss: "now" is EXACTLY the old zero-margin eligibility instant + // (bufferMs past a round line). With the epsilon margin the scheduler must NOT target this + // instant (delay 0) and must NOT skip a whole cycle — it targets epsilon later. + long roundTimeMs = discovery.roundTimeMills; + long bufferMs = discovery.bufferMillis; + long epsilon = discovery.getAlignedDelayEpsilonMillis(); + + AtomicLong mockTime = new AtomicLong(); + EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() { + @Override + public long currentTime() { + return mockTime.get(); + } + }); + + try { + long oldEligibilityInstant = roundTimeMs * 5 + bufferMs; // where a bare tick used to land + mockTime.set(oldEligibilityInstant); + long delay = discovery.computeAlignedInitialDelay(); + + assertEquals("Should wake epsilon after the eligibility instant, not on it", epsilon, delay); + long targetTick = mockTime.get() + delay; + assertEquals("Target must be on the epsilon-shifted grid", 0, + (targetTick - bufferMs - epsilon) % roundTimeMs); } finally { EnvironmentEdgeManager.reset(); } @@ -2421,6 +2637,14 @@ public ScheduledExecutorService getScheduler() { return super.scheduler; } + public void setScheduler(ScheduledExecutorService schedulerToUse) { + this.scheduler = schedulerToUse; + } + + public void setRunning(boolean running) { + this.isRunning = running; + } + private Boolean mockShouldProcessInProgressDirectory = null; @Override