From 72951b0537f6d1a63631b2ec7995968d859a1bb6 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 13 Aug 2026 01:19:41 +0530 Subject: [PATCH 1/2] 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 From 5b550f98819efdc7e84f714c22cb335638ed2d22 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Sat, 15 Aug 2026 18:28:22 +0530 Subject: [PATCH 2/2] PHOENIX-7978 Harden replay poll rescheduling, error handling, and epsilon validation - Self-heal on mid-chain reschedule failure: if scheduleNextReplay() throws after a healthy start, mark the discovery not-running and shut the executor down so the ReplicationLogReplayService supervisor rebuilds it, instead of leaving isRunning=true with an idle executor that wedges polling until an RS restart. - Guard against re-selecting the just-processed grid point: track lastAlignedTargetMillis and advance one round when a wake lands on/before it; derive the delay and the absolute target from a single clock read. - Validate the aligned-delay epsilon: clamp non-numeric / out-of-range [0, roundTimeMills) values to the default with a one-shot WARN rather than throwing (which would wedge the group in the supervisor-retry loop). - Split the replay() catch: Exception logs and continues the chain; Error is logged (it would otherwise vanish into the discarded Future), tears the chain down, and is rethrown; the reschedule moved out of the finally block. - Downgrade the per-cycle "Scheduling next replay" INFO line to DEBUG. --- .../replication/ReplicationLogDiscovery.java | 149 +++++++++++++++--- .../ReplicationLogDiscoveryTest.java | 107 +++++++++++-- 2 files changed, 219 insertions(+), 37 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 53af2a7e26b..21e1117b87d 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 @@ -27,6 +27,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import javax.annotation.concurrent.GuardedBy; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; @@ -133,6 +134,21 @@ public abstract class ReplicationLogDiscovery { protected MetricsReplicationLogDiscovery metrics; protected long roundTimeMills; protected long bufferMillis; + /** + * Wall-clock instant (ms) of the round-eligibility grid point the most recently scheduled cycle + * targets, or {@link Long#MIN_VALUE} before the first schedule of the current generation. Used by + * {@link #scheduleNextReplay()} to guarantee each reschedule advances to a grid point strictly + * after the previous one, so a wake landing exactly on (delay 0) or slightly before (nanoTime + * skew) the boundary it just processed does not re-select the same grid point and run a redundant + * cycle. Reset in {@link #start()} so a restarted generation re-anchors from scratch. + */ + @GuardedBy("this") + protected long lastAlignedTargetMillis = Long.MIN_VALUE; + /** + * One-shot guard so a misconfigured {@link #REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY} logs its + * fall-back WARN once rather than every scheduling cycle (the epsilon is read live per cycle). + */ + private final AtomicBoolean warnedInvalidEpsilon = new AtomicBoolean(false); public ReplicationLogDiscovery(final ReplicationLogTracker replicationLogTracker) { this.replicationLogTracker = replicationLogTracker; @@ -178,6 +194,9 @@ public void start() throws IOException { executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); scheduler = executor; isRunning = true; + // Re-anchor the aligned-target guard so this fresh generation schedules from the next grid + // point rather than being constrained by a target left over from a previous start()/stop(). + lastAlignedTargetMillis = Long.MIN_VALUE; try { scheduleNextReplay(); } catch (RuntimeException | Error e) { @@ -235,18 +254,33 @@ public void stop() { */ @GuardedBy("this") protected void scheduleNextReplay() { - long delayMs = computeAlignedInitialDelay(); + long now = EnvironmentEdgeManager.currentTime(); + long delayMs = computeAlignedInitialDelay(now); + // Guarantee the next fire targets a grid point strictly after the one the previous cycle + // targeted. A wake landing exactly on (delayMs == 0) or slightly before (nanoTime skew, small + // positive delay) the boundary we just processed would otherwise re-select the same grid point + // and run a redundant cycle before the clock advances past it; bump one full round in that + // case. targetMillis is derived from the same clock read as delayMs, so the two never skew. + long targetMillis = now + delayMs; + if (targetMillis <= lastAlignedTargetMillis) { + delayMs += roundTimeMills; + targetMillis += roundTimeMills; + } + lastAlignedTargetMillis = targetMillis; // 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); + LOG.debug("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. + * A recoverable {@link Exception} from {@link #replay()} is logged and the chain continues, so a + * single failed round does not break it. A fatal {@link Error} (OOM, stack overflow, linkage) is + * logged, tears the chain down (marking the service not-running so the supervisor can rebuild it), + * and is rethrown -- never rescheduled onto a potentially corrupted JVM. * 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"). @@ -258,24 +292,49 @@ protected void scheduleNextReplay() { protected void runReplayCycle(ScheduledExecutorService owner) { try { replay(); - } catch (Throwable t) { - LOG.error("Error during replay for haGroup: {}", haGroupName, t); - } finally { + } catch (Exception e) { + // Recoverable failure: log and fall through to reschedule so a single failed round does not + // break the self-rescheduling chain. + LOG.error("Error during replay for haGroup: {}", haGroupName, e); + } catch (Error e) { + // Fatal JVM condition (OutOfMemoryError, StackOverflowError, linkage failure). Do not swallow + // it and do not reschedule another replay on a potentially corrupted JVM. Log it first -- + // otherwise the executor's discarded Future would hide it entirely -- then tear the chain + // down exactly as a broken reschedule does below (mark the service not-running and shut this + // executor down so the ReplicationLogReplayService supervisor can rebuild a fresh one) and + // rethrow. Guarded on owner == scheduler so a stale cycle does not tear down a newer + // generation's scheduler. + LOG.error("Fatal error during replay for haGroup: {}; replay polling stopped, will be " + + "restarted by the replay service supervisor", haGroupName, e); 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); - } + if (owner == scheduler) { + isRunning = false; + owner.shutdown(); + } + } + throw e; + } + // Reached only on normal completion or a caught (recoverable) Exception -- never after a fatal + // Error, which propagates out above without rescheduling. + 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) { + // scheduleNextReplay() failed unexpectedly (something other than the benign + // RejectedExecutionException handled above). The poll chain is already broken, so mark + // the service not-running and shut down this now-idle executor. The + // ReplicationLogReplayService supervisor re-invokes start() on its fixed-rate cadence + // and, seeing isRunning==false, rebuilds a fresh executor -- self-healing instead of + // silently wedging with isRunning==true (which would keep isRunning() reporting healthy + // while nothing polls, and make every later start() no-op as "already running"). + LOG.error("Failed to schedule next replay for haGroup: {}; replay polling stopped, " + + "will be restarted by the replay service supervisor", haGroupName, t); + isRunning = false; + owner.shutdown(); } } } @@ -606,7 +665,17 @@ public double getWaitingBufferPercentage() { * @return the initial delay in milliseconds until the next round-eligible tick */ protected long computeAlignedInitialDelay() { - long now = EnvironmentEdgeManager.currentTime(); + return computeAlignedInitialDelay(EnvironmentEdgeManager.currentTime()); + } + + /** + * Overload of {@link #computeAlignedInitialDelay()} that aligns against a caller-supplied + * {@code now}, so a caller can derive both the delay and the absolute target grid instant + * ({@code now + delay}) from a single clock read without the two skewing across reads. + * @param now the reference wall-clock instant in milliseconds + * @return the delay in milliseconds until the next round-eligible tick at or after {@code now} + */ + protected long computeAlignedInitialDelay(long now) { // 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(); @@ -625,13 +694,41 @@ public int getInProgressFileMinAgeSeconds() { } /** - * 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}). + * Returns the epsilon margin (milliseconds) added to the aligned scheduler wake instant. Guards + * against a misconfigured value: a non-numeric string, a negative value (which would move wakes + * before eligibility and reintroduce missed rounds), or a value {@code >= roundTimeMills} + * (which wraps through {@link Math#floorMod} and no longer represents the documented "epsilon + * after the boundary") all fall back to {@link #DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS} with a + * one-shot WARN. Riding over the misconfiguration keeps replay polling on a safe default rather + * than wedging the group, while the WARN makes the bad config visible. + * @return the epsilon margin in milliseconds, always within {@code [0, roundTimeMills)}. */ public long getAlignedDelayEpsilonMillis() { - return conf.getLong(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY, - DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS); + long epsilon; + try { + epsilon = conf.getLong(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY, + DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS); + } catch (NumberFormatException e) { + // Hadoop's getLong throws (rather than returning the default) when the key is present but + // not parseable as a number. + warnInvalidEpsilon("non-numeric value \"" + + conf.get(REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY) + "\""); + return DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS; + } + if (epsilon < 0 || epsilon >= roundTimeMills) { + warnInvalidEpsilon(epsilon + "ms"); + return DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS; + } + return epsilon; + } + + private void warnInvalidEpsilon(String badValueDescription) { + if (warnedInvalidEpsilon.compareAndSet(false, true)) { + LOG.warn( + "Invalid {} ({}) for haGroup: {}; must be within [0, {}). Falling back to default {}ms.", + REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY, badValueDescription, haGroupName, + roundTimeMills, DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS); + } } public ReplicationLogTracker getReplicationLogFileTracker() { 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 5851d360fff..db9ae36fc23 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 @@ -23,6 +23,7 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import static org.mockito.Mockito.*; import java.io.IOException; @@ -216,36 +217,55 @@ public void testRunReplayCycleReschedulesAfterReplayThrows() throws IOException } @Test - public void testRunReplayCycleReschedulesAfterReplayThrowsError() throws IOException { + public void testRunReplayCyclePropagatesErrorAndStopsChain() throws IOException { + // setUp() stubs isRunning() to always return true; read the real field for this lifecycle test. + Mockito.doCallRealMethod().when(discovery).isRunning(); 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(); + // A fatal Error (OOME/StackOverflow/linkage) from replay() must NOT be swallowed and must NOT + // reschedule another replay on a possibly-corrupted JVM. It is logged and rethrown, and the + // chain is torn down (isRunning=false + executor shutdown) so the supervisor rebuilds it. + Error boom = new Error("boom"); + doThrow(boom).when(discovery).replay(); doNothing().when(discovery).scheduleNextReplay(); - discovery.runReplayCycle(owner); // must not propagate the Error + try { + discovery.runReplayCycle(owner); + fail("fatal Error from replay() must propagate out of runReplayCycle"); + } catch (Error e) { + assertSame(boom, e); + } verify(discovery, times(1)).replay(); - verify(discovery, times(1)).scheduleNextReplay(); + verify(discovery, never()).scheduleNextReplay(); + assertFalse("fatal Error must mark the service not-running for supervisor restart", + discovery.isRunning()); + verify(owner, times(1)).shutdown(); } @Test public void testRunReplayCycleDoesNotPropagateWhenRescheduleThrows() throws IOException { + // setUp() stubs isRunning() to always return true; read the real field for this lifecycle test. + Mockito.doCallRealMethod().when(discovery).isRunning(); 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. + // An unexpected failure in the reschedule path (e.g. computeAlignedInitialDelay -> getLong + // throwing NumberFormatException) must be caught, not swallowed by the executor into the + // discarded Future. The chain is already broken, so runReplayCycle marks the service + // not-running and shuts this executor down, letting the ReplicationLogReplayService supervisor + // restart it on its next tick -- instead of silently wedging 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(); + assertFalse("reschedule failure must mark the service not-running for supervisor restart", + discovery.isRunning()); + verify(owner, times(1)).shutdown(); } @Test @@ -267,7 +287,7 @@ public void testRunReplayCycleDoesNotRescheduleForStaleSchedulerGeneration() thr 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(); + doReturn(TimeUnit.HOURS.toMillis(1)).when(discovery).computeAlignedInitialDelay(anyLong()); doNothing().when(discovery).replay(); discovery.start(); // gen-1 @@ -292,7 +312,7 @@ public void testScheduleNextReplayUsesAlignedDelay() { ScheduledExecutorService mockScheduler = mock(ScheduledExecutorService.class); discovery.setScheduler(mockScheduler); long knownDelay = 1_234L; - doReturn(knownDelay).when(discovery).computeAlignedInitialDelay(); + doReturn(knownDelay).when(discovery).computeAlignedInitialDelay(anyLong()); discovery.scheduleNextReplay(); @@ -300,6 +320,36 @@ public void testScheduleNextReplayUsesAlignedDelay() { eq(TimeUnit.MILLISECONDS)); } + @Test + public void testScheduleNextReplayAdvancesPastLastTargetedGrid() { + // A wake that lands exactly on the boundary it just processed (aligned delay == 0) must not + // re-select the same grid point: the next reschedule at that same instant advances a full + // round instead of scheduling 0 again (which would run a redundant, no-op cycle immediately). + ScheduledExecutorService mockScheduler = mock(ScheduledExecutorService.class); + discovery.setScheduler(mockScheduler); + long roundTimeMs = discovery.roundTimeMills; + // Pin wall-clock exactly on an epsilon-shifted grid tick so computeAlignedInitialDelay == 0. + long onTick = 5 * roundTimeMs + discovery.bufferMillis + discovery.getAlignedDelayEpsilonMillis(); + AtomicLong mockTime = new AtomicLong(onTick); + EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() { + @Override + public long currentTime() { + return mockTime.get(); + } + }); + try { + discovery.scheduleNextReplay(); // on tick -> delay 0, targets grid point `onTick` + discovery.scheduleNextReplay(); // same instant -> must bump one full round, not schedule 0 + + verify(mockScheduler, times(1)).schedule(any(Runnable.class), eq(0L), + eq(TimeUnit.MILLISECONDS)); + verify(mockScheduler, times(1)).schedule(any(Runnable.class), eq(roundTimeMs), + eq(TimeUnit.MILLISECONDS)); + } finally { + EnvironmentEdgeManager.reset(); + } + } + @Test public void testComputeAlignedInitialDelay() { long roundTimeMs = discovery.roundTimeMills; @@ -493,6 +543,41 @@ public void testAlignedDelayEpsilonDefaultAndConfig() { conf.unset(ReplicationLogDiscovery.REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY); } + @Test + public void testAlignedDelayEpsilonInvalidFallsBackToDefault() { + long roundTimeMs = discovery.roundTimeMills; + String key = ReplicationLogDiscovery.REPLICATION_ALIGNED_DELAY_EPSILON_MILLIS_KEY; + long defaultEpsilon = ReplicationLogDiscovery.DEFAULT_ALIGNED_DELAY_EPSILON_MILLIS; + + // Valid boundary values pass through unchanged. + conf.setLong(key, 0L); + assertEquals("0 is in range and must pass through", 0L, + discovery.getAlignedDelayEpsilonMillis()); + conf.setLong(key, roundTimeMs - 1); + assertEquals("roundTimeMills - 1 is in range and must pass through", roundTimeMs - 1, + discovery.getAlignedDelayEpsilonMillis()); + + // Negative would move wakes before eligibility -> clamp to default. + conf.setLong(key, -1L); + assertEquals("negative epsilon must fall back to default", defaultEpsilon, + discovery.getAlignedDelayEpsilonMillis()); + + // >= roundTimeMills wraps through floorMod -> clamp to default (boundary and beyond). + conf.setLong(key, roundTimeMs); + assertEquals("epsilon == roundTimeMills must fall back to default", defaultEpsilon, + discovery.getAlignedDelayEpsilonMillis()); + conf.setLong(key, roundTimeMs + 5_000L); + assertEquals("epsilon > roundTimeMills must fall back to default", defaultEpsilon, + discovery.getAlignedDelayEpsilonMillis()); + + // Non-numeric value: Hadoop's getLong throws NumberFormatException -> clamp to default. + conf.set(key, "not-a-number"); + assertEquals("non-numeric epsilon must fall back to default", defaultEpsilon, + discovery.getAlignedDelayEpsilonMillis()); + + conf.unset(key); + } + @Test public void testComputeAlignedInitialDelayWakesEpsilonAfterEligibility() { // Reproduces the near-miss: "now" is EXACTLY the old zero-margin eligibility instant