You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by tk...@apache.org on 2026/09/17 22:49:29 UTC
(phoenix) branch PHOENIX-7562-feature-new updated: PHOENIX-8005 Make scheduled round-boundary rotation non-coalescible (#2620)
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 <kh...@gmail.com>
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.