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 888fa0f457 PHOENIX-8005 Make scheduled round-boundary rotation
non-coalescible (#2620)
888fa0f457 is described below
commit 888fa0f457c2f540f7aa5492b0d6acabc477f4c9
Author: tkhurana <[email protected]>
AuthorDate: Thu Sep 17 15:49:23 2026 -0700
PHOENIX-8005 Make scheduled round-boundary rotation non-coalescible (#2620)
* PHOENIX-8005 Make scheduled round-boundary rotation non-coalescible
The time-based rotation tick and size/fault rotations shared one
rotationRequested CAS gate, so a scheduled boundary tick firing while an
on-demand rotation held the gate could be coalesced away, leaving a round
boundary without its own .plog file.
Make coalescing asymmetric in requestRotation(boolean): on-demand callers
still CAS the gate (coalescing into any in-flight rotation, including a
scheduled one), while the scheduled tick force-holds the gate with set(true)
and always enqueues, so a boundary is never coalesced away. No new
synchronization state; LogRotationTask still clears the gate in its finally.
Adds testScheduledTickNotCoalescedWhenGateHeld and
testOnDemandCoalescesIntoScheduledRotation.
---
.../apache/phoenix/replication/ReplicationLog.java | 84 +++++++++++++--------
.../replication/ReplicationLogGroupTest.java | 85 ++++++++++++++++++++++
2 files changed, 139 insertions(+), 30 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
index a8ef6ee3c0..25fd8450e4 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLog.java
@@ -82,11 +82,11 @@ public class ReplicationLog {
// stranded. The waited-on condition is pendingWriter itself, so a spurious
notify is harmless.
private final Object rotationSignal = new Object();
private final AtomicBoolean closed = new AtomicBoolean(false);
- // Single gate for rotation submission. Set by requestRotation()'s CAS
before queuing a task,
- // cleared in LogRotationTask's finally. Both scheduled ticks and on-demand
callers go through
- // requestRotation(), so a request that arrives while a rotation is queued
or running is a
- // no-op — preventing the duplicate-writer bug where an in-flight scheduled
rotation and a
- // concurrent size-triggered request both stage writers and the second
closes the first.
+ // Single gate for rotation submission, cleared in LogRotationTask's
finally. On-demand
+ // (size/fault) callers CAS it, so a request arriving while a rotation is
queued or running
+ // coalesces — preventing the duplicate-writer bug where two requests both
stage writers and the
+ // second closes the first. The scheduled tick instead force-holds it
(set(true)) and always
+ // enqueues, so a round boundary is never coalesced away; see
requestRotation(boolean).
@VisibleForTesting
final AtomicBoolean rotationRequested = new AtomicBoolean(false);
private final ExecutorService closeExecutor;
@@ -170,9 +170,10 @@ public class ReplicationLog {
rotationExecutor = Executors.newSingleThreadScheduledExecutor(new
ThreadFactoryBuilder()
.setNameFormat("ReplicationLogRotation-" + logGroup.getHAGroupName() +
"-%d").setDaemon(true)
.build());
- // Scheduled ticks route through requestRotation() so they share the same
CAS gate as on-demand
- // size-triggered rotations — only one rotation can be queued or running
at a time.
- rotationExecutor.scheduleAtFixedRate(this::requestRotation, initialDelay,
rotationTimeMs,
+ // Scheduled ticks force the rotation gate (see requestRotation(boolean))
so a round boundary is
+ // never coalesced away by an in-flight size-triggered rotation; on-demand
rotations still
+ // coalesce, including into a scheduled one.
+ rotationExecutor.scheduleAtFixedRate(() -> requestRotation(true),
initialDelay, rotationTimeMs,
TimeUnit.MILLISECONDS);
LOG.info("Started rotation executor with initial delay {}ms and interval
{}ms", initialDelay,
rotationTimeMs);
@@ -240,34 +241,56 @@ public class ReplicationLog {
}
}
+ /** On-demand (size/fault) callers request a rotation that coalesces via the
gate. */
+ private boolean requestRotation() {
+ return requestRotation(false);
+ }
+
/**
- * Submits a {@link LogRotationTask} to the executor. The CAS gate ensures
only one rotation can
- * be queued or in flight at a time — if the flag is already set, a task is
already pending or
- * running. Both scheduled ticks and on-demand size-triggered callers go
through this method.
+ * Submits a {@link LogRotationTask} to the executor. The {@code
rotationRequested} gate coalesces
+ * rotations, but <b>asymmetrically</b> depending on the caller:
+ * <ul>
+ * <li><b>On-demand</b> (size/fault, {@code scheduled == false}): CAS the
gate; if a rotation is
+ * already queued or in flight the request coalesces into it (idempotent
"ensure a fresh writer" —
+ * a duplicate would just mint and immediately orphan-close a file). This
now also coalesces into
+ * an in-flight scheduled rotation, since that too is minting a fresh writer.
+ * <li><b>Scheduled tick</b> ({@code scheduled == true}): <i>force-hold</i>
the gate with
+ * {@code set(true)} and always enqueue. A round boundary is the placement
unit the round-sharded
+ * reader is organized around, so it must never be coalesced away by an
in-flight on-demand
+ * rotation. Holding the gate for the rotation's duration also makes
concurrent on-demand requests
+ * coalesce into it.
+ * </ul>
+ * The enqueued {@link LogRotationTask} clears the gate in its {@code
finally} regardless of which
+ * caller set it.
* <p>
* Suspends rotation during the in-sync cutover gate: while {@code
failoverPending} is set we skip
- * before the CAS so a tick never engages it. Minting a new file each round
would keep dropping
- * .plog files into the peer's shard directory and deadlock the standby's
failover check; the
- * current writer stays open so in-flight writes still land. Skipping ahead
of the CAS (rather
- * than inside {@link LogRotationTask#run()}) keeps the gate clear, so a
later tick resumes
- * rotation as soon as the flag clears on abort.
+ * before touching the gate so a tick never engages it. Minting a new file
each round would keep
+ * dropping .plog files into the peer's shard directory and deadlock the
standby's failover check;
+ * the current writer stays open so in-flight writes still land.
* @return {@code true} if a rotation is now queued or already in flight
(worth waiting for);
* {@code false} if rotation is suppressed this call (failover
pending, or the executor is
* shutting down) so no task will run.
*/
- private boolean requestRotation() {
+ @VisibleForTesting
+ boolean requestRotation(boolean scheduled) {
if (logGroup.isFailoverPending()) {
LOG.info("HAGroup {} rotation suspended: failover pending", logGroup);
return false;
}
- if (rotationRequested.compareAndSet(false, true)) {
- try {
- rotationExecutor.execute(new LogRotationTask());
- } catch (java.util.concurrent.RejectedExecutionException e) {
- LOG.info("Rotation executor shut down, skipping rotation", e);
- rotationRequested.set(false);
- return false;
- }
+ if (scheduled) {
+ // Force-hold rather than CAS: a round boundary is never coalesced away,
and holding the gate
+ // makes concurrent on-demand requests coalesce into this rotation.
+ rotationRequested.set(true);
+ } else if (!rotationRequested.compareAndSet(false, true)) {
+ // On-demand request coalesces into the rotation already queued or
running.
+ return true;
+ }
+ try {
+ rotationExecutor.execute(new LogRotationTask());
+ } catch (java.util.concurrent.RejectedExecutionException e) {
+ LOG.info("Rotation executor shut down, skipping rotation", e);
+ rotationRequested.set(false);
+ return false;
}
return true;
}
@@ -498,12 +521,13 @@ public class ReplicationLog {
/**
* Creates a new writer on a background thread and stages it in {@code
pendingWriter} for the
- * consumer thread to drain inside {@link #apply}. Submitted via {@link
#requestRotation()} from
- * both the scheduled tick and on-demand size-triggered callers.
+ * consumer thread to drain inside {@link #apply}. Submitted via {@link
#requestRotation(boolean)}
+ * from both the scheduled tick and on-demand size-triggered callers.
* <p>
- * {@code rotationRequested} stays set for the entire duration of run()
(cleared in finally), so
- * any request that arrives while this task is creating/staging a writer is
rejected by
- * {@link #requestRotation()}'s CAS and not duplicated.
+ * {@code rotationRequested} stays set for the entire duration of run()
(cleared unconditionally
+ * in finally), so an on-demand request that arrives while this task is
creating/staging a writer
+ * is rejected by {@link #requestRotation(boolean)}'s CAS and coalesced —
regardless of whether
+ * this task was enqueued by an on-demand caller or a scheduled tick.
*/
protected class LogRotationTask implements Runnable {
@Override
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
index b59e4cb96d..3dfde73c26 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogGroupTest.java
@@ -2053,6 +2053,91 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
writerAfterScheduledTick != writerAfterSizeRotation);
}
+ /**
+ * The scheduled tick is never coalesced away. With the gate already held
(as an on-demand
+ * size/fault rotation would hold it), a scheduled tick must still enqueue
its own rotation task
+ * so the round boundary gets its own writer. Holds the gate with a bare
flag set and directly
+ * drives {@code requestRotation(true)} to isolate the gate decision from
executor scheduling
+ * order; the genuine in-flight overlap is covered by
+ * {@link #testOnDemandCoalescesIntoScheduledRotation}.
+ */
+ @Test
+ public void testScheduledTickNotCoalescedWhenGateHeld() throws Exception {
+ ReplicationLog activeLog = logGroup.getActiveLog();
+
+ AtomicInteger writersCreated = new AtomicInteger(0);
+ doAnswer(invocation -> {
+ LogFileWriter w = (LogFileWriter) invocation.callRealMethod();
+ writersCreated.incrementAndGet();
+ return w;
+ }).when(activeLog).createNewWriter();
+
+ // Hold the gate as an on-demand size/fault rotation would, without a task
actually running.
+ activeLog.rotationRequested.set(true);
+
+ // Scheduled tick fires while the gate is held: it force-holds and always
enqueues.
+ assertTrue("Scheduled tick must report a rotation queued",
activeLog.requestRotation(true));
+
+ // The enqueued task runs on the rotation executor, creates a writer, and
clears the gate in its
+ // finally. Wait on the gate clearing rather than the writer count:
createNewWriter (which bumps
+ // the counter) runs before the finally, so polling the count alone could
observe the flag still
+ // held mid-run.
+ long deadline = System.currentTimeMillis() + 5000;
+ while (activeLog.rotationRequested.get() && System.currentTimeMillis() <
deadline) {
+ Thread.sleep(20);
+ }
+ assertFalse("Scheduled rotation task must clear the gate in finally",
+ activeLog.rotationRequested.get());
+ assertEquals("Scheduled tick must create its own writer even when the gate
was held", 1,
+ writersCreated.get());
+ }
+
+ /**
+ * On-demand (size/fault) rotations coalesce into an in-flight rotation of
any kind, including a
+ * scheduled one. With a scheduled rotation <i>genuinely in flight</i> — its
+ * {@link LogRotationTask} running on the executor and blocked inside {@code
createNewWriter}
+ * while holding the gate — an on-demand request must not enqueue a second
task; it coalesces and
+ * returns true (worth waiting for the writer the scheduled rotation will
stage).
+ */
+ @Test
+ public void testOnDemandCoalescesIntoScheduledRotation() throws Exception {
+ ReplicationLog activeLog = logGroup.getActiveLog();
+
+ CountDownLatch writerEntered = new CountDownLatch(1);
+ CountDownLatch releaseWriter = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ writerEntered.countDown();
+ assertTrue("Scheduled rotation was not released in time",
+ releaseWriter.await(5, TimeUnit.SECONDS));
+ return invocation.callRealMethod();
+ }).when(activeLog).createNewWriter();
+ // Drop the init-time createNewWriter call so the verify below counts only
test-driven
+ // rotations.
+ Mockito.clearInvocations(activeLog);
+
+ // Put a scheduled rotation genuinely in flight: it enqueues a task that
force-holds the gate
+ // and
+ // blocks inside createNewWriter until we release it.
+ assertTrue("Scheduled tick must enqueue a rotation",
activeLog.requestRotation(true));
+ assertTrue("Scheduled rotation task must reach createNewWriter",
+ writerEntered.await(5, TimeUnit.SECONDS));
+
+ // On-demand request arrives while the scheduled rotation holds the gate:
it must coalesce (no
+ // second task) and report the in-flight rotation is worth waiting for.
+ assertTrue("On-demand request must report the in-flight rotation is worth
waiting for",
+ activeLog.requestRotation(false));
+ assertTrue("Gate stays held by the in-flight scheduled rotation",
+ activeLog.rotationRequested.get());
+
+ // Release the scheduled rotation; it stages its writer and clears the
gate in finally. A
+ // coalescing regression would have enqueued a second task that the
single-thread executor runs
+ // next; after() waits out that window, then asserts exactly one writer
was ever created.
+ releaseWriter.countDown();
+ verify(activeLog, Mockito.after(500).times(1)).createNewWriter();
+ assertFalse("Scheduled rotation must clear the gate in finally",
+ activeLog.rotationRequested.get());
+ }
+
/**
* Tests that the rotation executor fires at the round boundary by verifying
that after waiting
* for slightly more than a round, a rotation has occurred.
