This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new 73037634d9 PHOENIX-7978 Harden replay/forward poll scheduling against
wall-clock… (#2598)
73037634d9 is described below
commit 73037634d99cf75ff9436f66b686b9978a880f3c
Author: Himanshu Gwalani <[email protected]>
AuthorDate: Thu Aug 20 04:39:00 2026 +0530
PHOENIX-7978 Harden replay/forward poll scheduling against wall-clock…
(#2598)
* 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.
* 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.
---------
Co-authored-by: Himanshu Gwalani <[email protected]>
---
.../replication/ReplicationLogDiscovery.java | 251 +++++++++++++--
.../ReplicationLogDiscoveryReplayTestIT.java | 29 --
.../replication/ReplicationLogDiscoveryTest.java | 353 +++++++++++++++++++--
3 files changed, 550 insertions(+), 83 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 638d95280f..21e1117b87 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,13 @@ import java.util.HashMap;
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 java.util.concurrent.atomic.AtomicBoolean;
+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,15 +102,53 @@ 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;
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;
@@ -132,9 +173,10 @@ public abstract class ReplicationLogDiscovery {
}
/**
- * 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 +185,29 @@ public abstract class ReplicationLogDiscovery {
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;
+ // 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) {
+ // 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 +245,101 @@ public abstract class ReplicationLogDiscovery {
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 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.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.
+ * 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").
+ * @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 (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 (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();
+ }
+ }
+ }
+ }
+
/**
* 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 +628,6 @@ public abstract class ReplicationLogDiscovery {
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 +659,27 @@ public abstract class ReplicationLogDiscovery {
* 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;
+ 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();
+ long elapsed = Math.floorMod(now - anchor, roundTimeMills);
return (elapsed == 0) ? 0 : roundTimeMills - elapsed;
}
@@ -544,6 +693,44 @@ public abstract class ReplicationLogDiscovery {
DEFAULT_IN_PROGRESS_FILE_MIN_AGE_SECONDS);
}
+ /**
+ * 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
+ * <em>before</em> 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() {
+ 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() {
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 98209280f4..e65525d284 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 class ReplicationLogDiscoveryReplayTestIT extends
HABaseIT {
"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 4af410e56e..db9ae36fc2 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.assertNotNull;
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;
@@ -139,10 +140,6 @@ public class ReplicationLogDiscoveryTest {
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 +158,208 @@ public class ReplicationLogDiscoveryTest {
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 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);
+ // 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();
+
+ 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, 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();
+ // 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
+ 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(anyLong());
+ 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(anyLong());
+
+ discovery.scheduleNextReplay();
+
+ verify(mockScheduler, times(1)).schedule(any(Runnable.class),
eq(knownDelay),
+ 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;
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 +384,10 @@ public class ReplicationLogDiscoveryTest {
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 +395,37 @@ public class ReplicationLogDiscoveryTest {
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 +445,12 @@ public class ReplicationLogDiscoveryTest {
});
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 +472,22 @@ public class ReplicationLogDiscoveryTest {
});
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 +509,101 @@ public class ReplicationLogDiscoveryTest {
});
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 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
+ // (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 +2722,14 @@ public class ReplicationLogDiscoveryTest {
return super.scheduler;
}
+ public void setScheduler(ScheduledExecutorService schedulerToUse) {
+ this.scheduler = schedulerToUse;
+ }
+
+ public void setRunning(boolean running) {
+ this.isRunning = running;
+ }
+
private Boolean mockShouldProcessInProgressDirectory = null;
@Override
