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 bea48ac8d0 PHOENIX-7988 Idle co-active RegionServer stays wedged in
STORE_AND_FORWARD, pinning the group at ACTIVE_NOT_IN_SYNC (#2600)
bea48ac8d0 is described below
commit bea48ac8d08ab347219bf0f73b12e60fea0b014c
Author: tkhurana <[email protected]>
AuthorDate: Mon Aug 17 09:59:10 2026 -0700
PHOENIX-7988 Idle co-active RegionServer stays wedged in STORE_AND_FORWARD,
pinning the group at ACTIVE_NOT_IN_SYNC (#2600)
A co-active RS in STORE_AND_FORWARD promotes back to SYNC_AND_FORWARD via
processNoMoreRoundsLeft only when the caught-up guard passes. That guard
scanned
the next round's shard, which is a shared directory holding every co-active
RS's
live OPENFORWRITE rotation writer, so an idle or out-claimed RS never saw it
empty and stayed pinned in STORE_AND_FORWARD, wedging the group at
ACTIVE_NOT_IN_SYNC indefinitely.
Split the guard: the RS's own mode flip is gated only on an empty
in-progress
directory (the per-RS forward-health signal), while the shared in-sync
status
claim keeps the full caught-up guard. The mode flip is self-validating since
SyncAndForwardModeImpl.onEnter must reach the peer, so an optimistic
promotion
against a dead peer bounces back to STORE_AND_FORWARD.
Adds a VisibleForTesting tracker-injecting constructor and three focused
tests
covering the wedge fix, the peer-down backlog case (no promotion), and the
fully-caught-up case.
---
.../ReplicationLogDiscoveryForwarder.java | 63 ++++++-----
.../ReplicationLogDiscoveryForwarderTest.java | 119 +++++++++++++++++++++
2 files changed, 158 insertions(+), 24 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
index aac363add4..eeaf6b757d 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
@@ -73,7 +73,12 @@ public class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
}
public ReplicationLogDiscoveryForwarder(ReplicationLogGroup logGroup) {
- super(createLogTracker(logGroup));
+ this(logGroup, createLogTracker(logGroup));
+ }
+
+ @VisibleForTesting
+ ReplicationLogDiscoveryForwarder(ReplicationLogGroup logGroup,
ReplicationLogTracker tracker) {
+ super(tracker);
this.logGroup = logGroup;
this.copyThroughputThresholdBytesPerMs = conf.getDouble(
REPLICATION_LOG_COPY_THROUGHPUT_BYTES_PER_MS_KEY,
DEFAULT_LOG_COPY_THROUGHPUT_BYTES_PER_MS);
@@ -156,32 +161,42 @@ public class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
@Override
protected void processNoMoreRoundsLeft() throws IOException {
- // check if we are caught up so that we can transition to SYNC state
- // we are caught up when there are no files currently in the out progress
directory
- // and no new files exist for ongoing round
+ // A non-empty in-progress directory means this RS has claimed-but-stuck
files, so its forward
+ // path to the peer is unhealthy: neither promote its own mode nor claim
the group is in sync.
+ if (!replicationLogTracker.getInProgressFiles().isEmpty()) {
+ LOG.info("In-progress directory not empty for {}, skipping mode
promotion and sync claim",
+ logGroup);
+ return;
+ }
+
+ // No stuck files, so promote this RS's own mode on that signal alone.
Gating on the next
+ // round's
+ // shard would pin an idle RS forever: it is a shared directory holding
every co-active RS's
+ // live
+ // rotation writer. The flip is self-validating —
SyncAndForwardModeImpl.onEnter must reach the
+ // peer, so a bad promotion bounces back to STORE_AND_FORWARD.
+ logGroup.checkAndSetModeAndNotify(STORE_AND_FORWARD, SYNC_AND_FORWARD);
+
+ // The shared in-sync claim additionally requires no new files for the
ongoing round.
if (
- replicationLogTracker.getInProgressFiles().isEmpty()
- && replicationLogTracker.getNewFilesForRound(replicationLogTracker
-
.getReplicationShardDirectoryManager().getNextRound(getLastRoundProcessed())).isEmpty()
+ !replicationLogTracker.getNewFilesForRound(replicationLogTracker
+
.getReplicationShardDirectoryManager().getNextRound(getLastRoundProcessed())).isEmpty()
) {
- LOG.info("Processed all the replication log files for {}", logGroup);
- // if this RS is still in STORE_AND_FORWARD mode like when it didn't
process any file
- // move this RS to SYNC_AND_FORWARD
- logGroup.checkAndSetModeAndNotify(STORE_AND_FORWARD, SYNC_AND_FORWARD);
-
- if (syncUpdateTS <= EnvironmentEdgeManager.currentTimeMillis()) {
- try {
- long waitTime = logGroup.setHAGroupStatusToSync();
- if (waitTime != 0) {
- syncUpdateTS = EnvironmentEdgeManager.currentTimeMillis() +
waitTime;
- LOG.info("HAGroup {} will try to update HA state to sync at {}",
logGroup,
- syncUpdateTS);
- } else {
- LOG.info("HAGroup {} updated HA state to SYNC", logGroup);
- }
- } catch (Exception e) {
- LOG.info("Could not update status to sync for {}", logGroup, e);
+ LOG.info("New files present for the next round for {}, skipping sync
claim", logGroup);
+ return;
+ }
+ LOG.info("Processed all the replication log files for {}", logGroup);
+ if (syncUpdateTS <= EnvironmentEdgeManager.currentTimeMillis()) {
+ try {
+ long waitTime = logGroup.setHAGroupStatusToSync();
+ if (waitTime != 0) {
+ syncUpdateTS = EnvironmentEdgeManager.currentTimeMillis() + waitTime;
+ LOG.info("HAGroup {} will try to update HA state to sync at {}",
logGroup, syncUpdateTS);
+ } else {
+ LOG.info("HAGroup {} updated HA state to SYNC", logGroup);
}
+ } catch (Exception e) {
+ LOG.info("Could not update status to sync for {}", logGroup, e);
}
}
}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java
index 269e67605b..725a2f0662 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java
@@ -19,20 +19,27 @@ package org.apache.phoenix.replication;
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.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
@@ -480,4 +487,116 @@ public class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
}
assertEquals(SYNC, logGroup.getMode());
}
+
+ /**
+ * Builds a forwarder over a mock tracker so processNoMoreRoundsLeft can be
driven directly with a
+ * controlled in-progress / next-round file state.
+ */
+ private ReplicationLogDiscoveryForwarder
forwarderWithMockTracker(ReplicationLogTracker tracker) {
+ ReplicationShardDirectoryManager shardManager =
mock(ReplicationShardDirectoryManager.class);
+ // stubs read by the ReplicationLogDiscovery constructor
+ doReturn(conf).when(tracker).getConf();
+ doReturn(haGroupName).when(tracker).getHaGroupName();
+ doReturn(shardManager).when(tracker).getReplicationShardDirectoryManager();
+ doReturn(20).when(shardManager).getReplicationRoundDurationSeconds();
+ doReturn(new ReplicationRound(0,
20_000)).when(shardManager).getNextRound(any());
+ ReplicationLogDiscoveryForwarder forwarder =
+ new ReplicationLogDiscoveryForwarder(logGroup, tracker);
+ forwarder.setLastRoundProcessed(new ReplicationRound(0, 20_000));
+ return forwarder;
+ }
+
+ /**
+ * The mode flip to SYNC_AND_FORWARD is gated only on an empty in-progress
directory, not on the
+ * next round's shard scan. An idle RS whose own (and its peers') live
rotation writer keeps the
+ * next-round scan non-empty must still promote its own mode; otherwise it
stays pinned in
+ * STORE_AND_FORWARD and wedges the group at ACTIVE_NOT_IN_SYNC.
+ */
+ @Test
+ public void testModeFlipsWhenInProgressEmptyEvenIfNextRoundHasFiles() throws
Exception {
+ ReplicationLogTracker tracker = mock(ReplicationLogTracker.class);
+ doReturn(Collections.emptyList()).when(tracker).getInProgressFiles();
+ // next round is non-empty (a live rotation writer masquerading as pending
work)
+ doReturn(Collections.singletonList(new
Path("out/shard/000/1_rs2.plog"))).when(tracker)
+ .getNewFilesForRound(any());
+ ReplicationLogDiscoveryForwarder forwarder =
forwarderWithMockTracker(tracker);
+
+ forwarder.processNoMoreRoundsLeft();
+
+ // mode promoted despite the non-empty next-round scan ...
+ verify(logGroup, times(1)).checkAndSetModeAndNotify(STORE_AND_FORWARD,
SYNC_AND_FORWARD);
+ assertEquals(SYNC_AND_FORWARD, logGroup.getMode());
+ // ... but the shared in-sync claim is withheld because the full guard is
not satisfied
+ verify(logGroup, never()).setHAGroupStatusToSync();
+ }
+
+ /**
+ * A non-empty in-progress directory means this RS has claimed-but-stuck
files (its forward path
+ * is unhealthy), so it must neither promote its own mode nor claim the
group is in sync.
+ */
+ @Test
+ public void testNoModeFlipWhenInProgressNotEmpty() throws Exception {
+ ReplicationLogTracker tracker = mock(ReplicationLogTracker.class);
+ doReturn(Collections.singletonList(new
Path("out_progress/1_rs2_uuid_2.plog"))).when(tracker)
+ .getInProgressFiles();
+ doReturn(Collections.emptyList()).when(tracker).getNewFilesForRound(any());
+ ReplicationLogDiscoveryForwarder forwarder =
forwarderWithMockTracker(tracker);
+
+ forwarder.processNoMoreRoundsLeft();
+
+ verify(logGroup, never()).checkAndSetModeAndNotify(STORE_AND_FORWARD,
SYNC_AND_FORWARD);
+ assertEquals(STORE_AND_FORWARD, logGroup.getMode());
+ verify(logGroup, never()).setHAGroupStatusToSync();
+ }
+
+ /**
+ * Fully caught up: in-progress empty and no new files for the ongoing
round. Both the mode flip
+ * and the shared in-sync claim fire.
+ */
+ @Test
+ public void testStatusFlipsWhenFullyCaughtUp() throws Exception {
+ ReplicationLogTracker tracker = mock(ReplicationLogTracker.class);
+ doReturn(Collections.emptyList()).when(tracker).getInProgressFiles();
+ doReturn(Collections.emptyList()).when(tracker).getNewFilesForRound(any());
+ doReturn(0L).when(logGroup).setHAGroupStatusToSync();
+ ReplicationLogDiscoveryForwarder forwarder =
forwarderWithMockTracker(tracker);
+
+ forwarder.processNoMoreRoundsLeft();
+
+ verify(logGroup, times(1)).checkAndSetModeAndNotify(STORE_AND_FORWARD,
SYNC_AND_FORWARD);
+ assertEquals(SYNC_AND_FORWARD, logGroup.getMode());
+ verify(logGroup, times(1)).setHAGroupStatusToSync();
+ }
+
+ /**
+ * The mode flip is self-validating: promoting to SYNC_AND_FORWARD drives
+ * SyncAndForwardModeImpl.onEnter, which must reach the peer. If the peer is
unreachable the flip
+ * bounces back to STORE_AND_FORWARD, so an optimistic promotion against a
dead peer does not
+ * leave the group stuck advertising a healthy forward path. In-progress is
empty (so the
+ * promotion fires) but getOrCreatePeerShardManager always throws (dead
peer).
+ */
+ @Test
+ public void testModeFlipBouncesBackWhenPeerUnreachable() throws Exception {
+ ReplicationLogTracker tracker = mock(ReplicationLogTracker.class);
+ doReturn(Collections.emptyList()).when(tracker).getInProgressFiles();
+ doReturn(Collections.emptyList()).when(tracker).getNewFilesForRound(any());
+ // dead peer: onEnter for SYNC_AND_FORWARD can never reach the peer
+ doThrow(new IOException("Peer namenode unavailable")).when(logGroup)
+ .getOrCreatePeerShardManager();
+ ReplicationLogDiscoveryForwarder forwarder =
forwarderWithMockTracker(tracker);
+
+ forwarder.processNoMoreRoundsLeft();
+
+ // the promotion is attempted ...
+ verify(logGroup, times(1)).checkAndSetModeAndNotify(STORE_AND_FORWARD,
SYNC_AND_FORWARD);
+ // ... but onEnter's failed peer reach bounces the mode back
asynchronously on the disruptor
+ long deadline = EnvironmentEdgeManager.currentTimeMillis() + 120_000;
+ while (
+ logGroup.getMode() != STORE_AND_FORWARD
+ && EnvironmentEdgeManager.currentTimeMillis() < deadline
+ ) {
+ Thread.sleep(100);
+ }
+ assertEquals(STORE_AND_FORWARD, logGroup.getMode());
+ }
}
