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 daaf9b7caf PHOENIX-8029 Make the replication mode-switch notification
non-blocking (#2641)
daaf9b7caf is described below
commit daaf9b7caf309fb31d8dfabf9f6489fb579e7e0e
Author: tkhurana <[email protected]>
AuthorDate: Tue Oct 27 11:24:24 2026 -0700
PHOENIX-8029 Make the replication mode-switch notification non-blocking
(#2641)
checkAndSetModeAndNotify woke the event handler with a blocking sync(),
which waits up to the sync timeout and can abort the RS. It runs on the
ZK cache-event thread (HA state listeners) and the forwarder thread.
Publish a non-blocking MODE_CHECK marker with tryNext() instead, like the
rotation SWAP marker. The handler applies the switch on the marker only
when no syncs are pending and the current batch is empty; otherwise it
applies at the next sync point. The mode-switch tail of
processPendingSyncs is extracted into checkModeSwitch, which skips the
switch once the group is closed, so close()'s drain does not apply
switches made by other threads.
---
.../phoenix/replication/ReplicationLogGroup.java | 68 ++++++++++++--
.../replication/ReplicationLogGroupTest.java | 104 +++++++++++++++++++++
2 files changed, 163 insertions(+), 9 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
index 8f05dd0c4d..b1751f11c5 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogGroup.java
@@ -20,6 +20,7 @@ package org.apache.phoenix.replication;
import static org.apache.hadoop.hbase.HConstants.DEFAULT_ZK_SESSION_TIMEOUT;
import static org.apache.hadoop.hbase.HConstants.ZK_SESSION_TIMEOUT;
import static
org.apache.phoenix.replication.ReplicationLogGroup.LogEvent.EVENT_TYPE_DATA;
+import static
org.apache.phoenix.replication.ReplicationLogGroup.LogEvent.EVENT_TYPE_MODE_CHECK;
import static
org.apache.phoenix.replication.ReplicationLogGroup.LogEvent.EVENT_TYPE_SWAP;
import static
org.apache.phoenix.replication.ReplicationLogGroup.LogEvent.EVENT_TYPE_SYNC;
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.INIT;
@@ -418,6 +419,7 @@ public class ReplicationLogGroup {
public static final byte EVENT_TYPE_DATA = 0;
public static final byte EVENT_TYPE_SYNC = 1;
public static final byte EVENT_TYPE_SWAP = 2;
+ public static final byte EVENT_TYPE_MODE_CHECK = 3;
public void setValues(int type, Record record, CompletableFuture<Void>
syncFuture) {
this.type = type;
@@ -434,6 +436,8 @@ public class ReplicationLogGroup {
return "SYNC";
case EVENT_TYPE_SWAP:
return "SWAP";
+ case EVENT_TYPE_MODE_CHECK:
+ return "MODE_CHECK";
default:
return "UNKNOWN(" + type + ")";
}
@@ -799,18 +803,15 @@ public class ReplicationLogGroup {
}
/**
- * Conditionally switch replication mode and, if it changed, notify the
event handler with a sync
- * so the disruptor picks up the new mode.
+ * Conditionally switch replication mode and, if it changed, wake the event
handler so it applies
+ * the new mode even when idle. Never blocks, aborts or throws: it runs on
the ZK cache-event
+ * thread (state listeners) and the forwarder thread.
*/
protected boolean checkAndSetModeAndNotify(ReplicationMode expectedMode,
ReplicationMode newMode) {
boolean switched = checkAndSetMode(expectedMode, newMode);
if (switched) {
- try {
- sync();
- } catch (IOException e) {
- LOG.info("Failed to notify event handler for {}", this, e);
- }
+ publishModeCheckEvent();
}
return switched;
}
@@ -1044,6 +1045,30 @@ public class ReplicationLogGroup {
}
}
+ /**
+ * Publish a non-blocking mode-check marker so an idle consumer applies a
mode switch promptly.
+ * Uses {@code tryNext()} like {@link #publishSwapEvent()}: when the ring
buffer is full, the
+ * consumer is busy and the switch applies at the next sync point. Skipped
once closed:
+ * {@code checkModeSwitch} ignores the switch then anyway.
+ */
+ protected void publishModeCheckEvent() {
+ if (isClosed()) {
+ LOG.info("HAGroup {} closed, skipping mode check event publish", this);
+ return;
+ }
+ try {
+ long sequence = ringBuffer.tryNext();
+ try {
+ LogEvent event = ringBuffer.get(sequence);
+ event.setValues(EVENT_TYPE_MODE_CHECK, null, null);
+ } finally {
+ ringBuffer.publish(sequence);
+ }
+ } catch (InsufficientCapacityException e) {
+ LOG.info("HAGroup {} ring buffer full, skipping mode check event
publish", this);
+ }
+ }
+
/**
* Check if this ReplicationLogGroup is closed.
* @return true if closed, false otherwise
@@ -1347,6 +1372,12 @@ public class ReplicationLogGroup {
return eventHandler.getCurrentModeImpl().getReplicationLog();
}
+ /** Returns the ring buffer's free slots; equals its size once every
published event is done. */
+ @VisibleForTesting
+ long getRingBufferRemainingCapacity() {
+ return ringBuffer.remainingCapacity();
+ }
+
protected void setHAGroupStatusToStoreAndForward() throws Exception {
try {
haGroupStoreManager.setHAGroupStatusToStoreAndForward(haGroupName);
@@ -1510,6 +1541,18 @@ public class ReplicationLogGroup {
// Doing the mode check on sync points makes the implementation more
robust
// since we can guarantee that all unsynced appends have been flushed to
the
// replication log before we switch the replication mode
+ checkModeSwitch(sequence);
+ }
+
+ /**
+ * Applies a mode switched on the replication group by another thread, if
any. Callers must
+ * guarantee that no unsynced appends are outstanding. Once closed,
switches made by other
+ * threads are ignored, so the drain stays in the current mode unless a
write fails.
+ */
+ private void checkModeSwitch(long sequence) throws IOException {
+ if (isClosed()) {
+ return;
+ }
ReplicationMode newMode = getMode();
if (newMode != currentModeImpl.getMode()) {
// some other thread switched the mode on the replication group
@@ -1596,8 +1639,8 @@ public class ReplicationLogGroup {
}
/**
- * Processes a single event from the Disruptor ring buffer. This method
handles both data and
- * sync events.
+ * Processes a single event from the Disruptor ring buffer: data and sync
events, and the SWAP
+ * and MODE_CHECK wake-up markers.
* <p>
* For data events, it:
* <ol>
@@ -1649,6 +1692,13 @@ public class ReplicationLogGroup {
// closed before the reader's round buffer expires. No append/sync
to perform.
currentModeImpl.getReplicationLog().checkAndReplaceWriter(true);
break;
+ case EVENT_TYPE_MODE_CHECK:
+ // Wake-up marker from checkAndSetModeAndNotify. Switch now only
if nothing is
+ // outstanding; otherwise the switch applies at the next sync
point.
+ if (pendingSyncs.isEmpty() &&
currentModeImpl.log.getCurrentBatch().isEmpty()) {
+ checkModeSwitch(sequence);
+ }
+ break;
default:
throw new UnsupportedOperationException("Unknown event type: " +
event.type);
}
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 22bc3f1d8e..25dbf4b45f 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
@@ -20,6 +20,7 @@ package org.apache.phoenix.replication;
import static java.lang.Thread.sleep;
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.STORE_AND_FORWARD;
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC;
+import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC_AND_FORWARD;
import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -55,6 +56,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
@@ -888,6 +890,108 @@ public class ReplicationLogGroupTest extends
ReplicationLogBaseTest {
}
}
+ private void waitForHandlerMode(ReplicationLogGroup.ReplicationMode expected)
+ throws InterruptedException {
+ long deadline = System.currentTimeMillis() + 5000;
+ while (
+ logGroup.eventHandler.getCurrentModeImpl().getMode() != expected
+ && System.currentTimeMillis() < deadline
+ ) {
+ sleep(20);
+ }
+ assertEquals(expected,
logGroup.eventHandler.getCurrentModeImpl().getMode());
+ }
+
+ /**
+ * An idle group applies a mode switch from checkAndSetModeAndNotify on its
own: the mode-check
+ * marker wakes the consumer, and no sync() is issued.
+ */
+ @Test
+ public void testModeSwitchNotifyAppliesWhenIdle() throws Exception {
+ assertEquals(SYNC, logGroup.eventHandler.getCurrentModeImpl().getMode());
+ assertTrue(logGroup.checkAndSetModeAndNotify(SYNC, SYNC_AND_FORWARD));
+ waitForHandlerMode(SYNC_AND_FORWARD);
+ verify(logGroup, never()).sync();
+ }
+
+ private void waitForRingBufferDrained() throws InterruptedException {
+ long deadline = System.currentTimeMillis() + 5000;
+ while (
+ logGroup.getRingBufferRemainingCapacity() < TEST_RINGBUFFER_SIZE
+ && System.currentTimeMillis() < deadline
+ ) {
+ sleep(20);
+ }
+ assertEquals(TEST_RINGBUFFER_SIZE,
logGroup.getRingBufferRemainingCapacity());
+ }
+
+ /**
+ * A mode-check marker that finds unsynced appends in the current batch must
not switch the mode:
+ * the switch waits for the next sync, so the outstanding appends are synced
in the old mode.
+ */
+ @Test
+ public void testModeSwitchNotifyDefersWhileAppendsUnsynced() throws
Exception {
+ logGroup.append("TBLMODEDEFER", 1L, LogFileTestUtil.newPut("row", 1, 1));
+ waitForRingBufferDrained();
+
+ assertTrue(logGroup.checkAndSetModeAndNotify(SYNC, SYNC_AND_FORWARD));
+ waitForRingBufferDrained();
+ assertEquals("The marker must not switch with unsynced appends
outstanding", SYNC,
+ logGroup.eventHandler.getCurrentModeImpl().getMode());
+
+ logGroup.sync();
+ waitForHandlerMode(SYNC_AND_FORWARD);
+ }
+
+ /**
+ * While the consumer is busy with an outstanding sync,
checkAndSetModeAndNotify returns at once.
+ */
+ @Test
+ public void testModeSwitchNotifyDoesNotBlockOnBusyConsumer() throws
Exception {
+ // Headroom so a slow CI run cannot time out the held sync, and no
scheduled rotation can swap
+ // the spied writer mid-test.
+ conf.setLong(ReplicationLogGroup.REPLICATION_LOG_SYNC_TIMEOUT_KEY,
30_000L);
+ conf.setInt(PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY, 3600);
+ recreateLogGroup();
+ final Mutation put = LogFileTestUtil.newPut("row", 1, 1);
+ LogFileWriter writer = logGroup.getActiveLog().getWriter();
+ CountDownLatch release = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ release.await();
+ return invocation.callRealMethod();
+ }).when(writer).sync();
+
+ logGroup.append("TBLMODEBUSY", 1L, put);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> syncResult = executor.submit(() -> {
+ logGroup.sync();
+ return null;
+ });
+ verify(writer, timeout(5000)).sync();
+
+ long start = System.nanoTime();
+ assertTrue(logGroup.checkAndSetModeAndNotify(SYNC, SYNC_AND_FORWARD));
+ assertTrue("checkAndSetModeAndNotify must not block on a busy consumer",
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) < 500);
+
+ release.countDown();
+ syncResult.get(5, TimeUnit.SECONDS);
+ waitForHandlerMode(SYNC_AND_FORWARD);
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ /** On a fatal group, checkAndSetModeAndNotify returns normally and never
aborts. */
+ @Test
+ public void testModeSwitchNotifyOnFatalGroupDoesNotAbort() throws Exception {
+ logGroup.eventHandler.setFatalException(new IOException("Simulated
fatal"));
+ assertTrue(logGroup.checkAndSetModeAndNotify(SYNC, SYNC_AND_FORWARD));
+ verify(logGroup, never()).abort(anyString(), any());
+ }
+
/**
* Tests rotation during an in-flight batch. When a pending writer is staged
by the rotation task
* while appends are in flight, the sync's apply() drains the pending
writer, replays the 5
