From ae5f06082ffda1ddf063e279051909d70fc8e02a Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 16 Jul 2026 17:01:07 +0530 Subject: [PATCH 1/4] =?UTF-8?q?Add=20replay-side=20handling=20for=20direct?= =?UTF-8?q?=20DEGRADED=5FSTANDBY=20=E2=86=92=20STANDBY=5FTO=5FACTIVE=20tra?= =?UTF-8?q?nsition=20(ANISTS=20=E2=86=92=20AISTS)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../replication/ReplicationLogDiscovery.java | 26 +- .../reader/ReplicationLogDiscoveryReplay.java | 122 ++++++--- .../ReplicationLogDiscoveryReplayTestIT.java | 249 +++++++++++++++++- 3 files changed, 353 insertions(+), 44 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 ccd0e82caf4..83288cde44d 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 @@ -104,7 +104,7 @@ public abstract class ReplicationLogDiscovery { protected final ReplicationLogTracker replicationLogTracker; protected ScheduledExecutorService scheduler; protected volatile boolean isRunning = false; - protected ReplicationRound lastRoundProcessed; + protected volatile ReplicationRound lastRoundProcessed; protected MetricsReplicationLogDiscovery metrics; protected long roundTimeMills; protected long bufferMillis; @@ -382,16 +382,29 @@ private Optional processOneRandomFile(final List files) throws IOExc /** Creates a new metrics source for monitoring operations. */ protected abstract MetricsReplicationLogDiscovery createMetricsSource(); + /** + * Initializes lastRoundProcessed, sampling the current time up front to use as the no-files + * fallback (see {@link #initializeLastRoundProcessed(long)}). Sampling here - before the file + * scans - keeps the fallback anchored to when initialization began. + * @throws IOException if there's an error reading file timestamps + */ + protected void initializeLastRoundProcessed() throws IOException { + initializeLastRoundProcessed(EnvironmentEdgeManager.currentTime()); + } + /** * Initializes lastRoundProcessed based on minimum timestamp from 1. In-progress files (highest * priority) - indicates partially processed rounds 2. New files (medium priority) - indicates - * unprocessed rounds waiting to be replayed 3. Current time (fallback) - used when no files - * exist, starts from current time The minimum timestamp is converted to a replication round using + * unprocessed rounds waiting to be replayed 3. fallbackCurrentTime - used when no files exist, + * starts from the supplied time The minimum timestamp is converted to a replication round using * getReplicationRoundFromEndTime(), which rounds down to the nearest round boundary to ensure we * start from a complete round. + * @param fallbackCurrentTime the timestamp to start from when no files exist; sampled by the + * caller at the start of initialization rather than re-read here, so a slow init (or a + * state transition during init) cannot push the starting round forward * @throws IOException if there's an error reading file timestamps */ - protected void initializeLastRoundProcessed() throws IOException { + protected void initializeLastRoundProcessed(long fallbackCurrentTime) throws IOException { Optional minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles(); if (minTimestampFromInProgressFiles.isPresent()) { LOG.info( @@ -412,11 +425,10 @@ protected void initializeLastRoundProcessed() throws IOException { this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager() .getReplicationRoundFromEndTime(minTimestampFromNewFiles.get()); } else { - long currentTime = EnvironmentEdgeManager.currentTime(); LOG.info("Initializing lastRoundProcessed for haGroup: {} from current time {}", - haGroupName, currentTime); + haGroupName, fallbackCurrentTime); this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager() - .getReplicationRoundFromEndTime(currentTime); + .getReplicationRoundFromEndTime(fallbackCurrentTime); } } } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java index 0b0f7ef5177..b19cc988dd4 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java @@ -151,6 +151,13 @@ public void init() throws IOException { clusterType == ClusterType.LOCAL && HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(toState) ) { + // Direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE skips the STANDBY event that normally + // drives recovery. If we are DEGRADED, schedule the rewind so replay() re-syncs from + // lastRoundInSync before shouldTriggerFailover() (which gates on SYNC) can promote. + // compareAndSet, not set: a listener firing while already SYNC (healthy failover) or + // SYNCED_RECOVERY (rewind already pending) must not clobber a good state. + replicationReplayState.compareAndSet(ReplicationReplayState.DEGRADED, + ReplicationReplayState.SYNCED_RECOVERY); failoverPending.set(true); LOG.info( "Failover trigger detected for {}. replicationReplayState={}. " @@ -202,50 +209,51 @@ protected void processFile(Path path) throws IOException { } /** - * Initializes lastRoundProcessed and lastRoundInSync based on HA group state. For DEGRADED states - * (DEGRADED_STANDBY, DEGRADED_STANDBY_FOR_WRITER): - Sets replicationReplayState to DEGRADED - - * Initializes lastRoundProcessed from minimum of: in-progress files, new files, or current time - - * Initializes lastRoundInSync from minimum of: lastSyncStateTimeInMs (from HA Store) or minimum - * timestamp from IN and IN PROGRESS files - This ensures lastRoundInSync represents the last - * known good sync point before degradation For SYNC states (STANDBY): - Sets - * replicationReplayState to SYNC - Calls parent's initializeLastRoundProcessed() to initialize - * lastRoundProcessed - Sets lastRoundInSync equal to lastRoundProcessed (both are in sync) + * Initializes lastRoundProcessed and lastRoundInSync based on the persisted HA group state. + *
    + *
  • DEGRADED_STANDBY: sets replicationReplayState to DEGRADED and initializes both rounds from + * the last known good sync point via {@link #initLastRoundsFromLastSyncPoint(HAGroupStoreRecord)} + * (lastRoundInSync from the minimum of lastSyncStateTimeInMs and the file frontier, so it + * represents the last consistent point before degradation).
    • + *
  • STANDBY_TO_ACTIVE: a restart while already mid-failover (e.g. after a direct + * DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). Sets replicationReplayState to + * SYNCED_RECOVERY and initializes both rounds the same way, so the first replay() rewinds to + * lastRoundInSync before failover can promote.
    • + *
  • Other states (e.g. STANDBY): sets replicationReplayState to SYNC, delegates to the parent + * to initialize lastRoundProcessed, and sets lastRoundInSync equal to lastRoundProcessed.
    • + *
+ * When the state is STANDBY_TO_ACTIVE, failoverPending is also armed. * @throws IOException if there's an error reading HA group state or file timestamps */ @Override protected void initializeLastRoundProcessed() throws IOException { LOG.info("Initializing last round processed for haGroup: {}", haGroupName); + // Sample current time BEFORE reading the HA group state, so if the group transitions (e.g. + // SYNC -> DEGRADED_STANDBY) during init, the starting round still anchors to when init began + // rather than to a later point after the state read and file scans. + long frontierStartTime = EnvironmentEdgeManager.currentTime(); HAGroupStoreRecord haGroupStoreRecord = getHAGroupRecord(); - LOG.info("Found HA Group state during initialization as {} for haGroup: {}", - haGroupStoreRecord.getHAGroupState(), haGroupName); - if ( - HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupStoreRecord.getHAGroupState()) - ) { + HAGroupStoreRecord.HAGroupState haGroupState = haGroupStoreRecord.getHAGroupState(); + LOG.info("Found HA Group state during initialization as {} for haGroup: {}", haGroupState, + haGroupName); + if (HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupState)) { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.DEGRADED); - long minimumTimestampFromFiles = EnvironmentEdgeManager.currentTime(); - Optional minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles(); - Optional minTimestampFromNewFiles = getMinTimestampFromNewFiles(); - if (minTimestampFromInProgressFiles.isPresent()) { - LOG.info("Found minimum timestamp from IN PROGRESS files as {}", - minTimestampFromInProgressFiles.get()); - minimumTimestampFromFiles = - Math.min(minimumTimestampFromFiles, minTimestampFromInProgressFiles.get()); - } - if (minTimestampFromNewFiles.isPresent()) { - LOG.info("Found minimum timestamp from IN files as {}", minTimestampFromNewFiles.get()); - minimumTimestampFromFiles = - Math.min(minimumTimestampFromFiles, minTimestampFromNewFiles.get()); - } - this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager() - .getReplicationRoundFromEndTime(minimumTimestampFromFiles); - this.lastRoundInSync = - replicationLogTracker.getReplicationShardDirectoryManager().getReplicationRoundFromEndTime( - Math.min(haGroupStoreRecord.getLastSyncStateTimeInMs(), minimumTimestampFromFiles)); + initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime); + } else if (HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) { + // Restarted while already in STANDBY_TO_ACTIVE (e.g. an RS bounce after a direct + // DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). No listener fires on a fresh process, so + // initialize as if recovering: lastRoundInSync at the last good sync point, lastRoundProcessed + // from the file frontier, and state SYNCED_RECOVERY so the first replay() rewinds before + // shouldTriggerFailover() can promote. Conservative by design: if we actually came from a + // healthy STANDBY, lastSyncStateTimeInMs ~= the file frontier and the rewind is a no-op. + replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, + ReplicationReplayState.SYNCED_RECOVERY); + initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime); } else { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.SYNC); - super.initializeLastRoundProcessed(); + super.initializeLastRoundProcessed(frontierStartTime); this.lastRoundInSync = new ReplicationRound(lastRoundProcessed.getStartTime(), lastRoundProcessed.getEndTime()); } @@ -255,13 +263,55 @@ protected void initializeLastRoundProcessed() throws IOException { lastRoundProcessed, lastRoundInSync, replicationReplayState); // Update the failoverPending variable during initialization - if ( - HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupStoreRecord.getHAGroupState()) - ) { + if (HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) { failoverPending.compareAndSet(false, true); } } + /** + * Initializes {@link #lastRoundProcessed} and {@link #lastRoundInSync} from the last known good + * sync point. Used by the DEGRADED_STANDBY and STANDBY_TO_ACTIVE branches, which both need a + * rewind to the last consistent point. lastRoundProcessed is derived from the minimum timestamp + * across IN-PROGRESS and IN files (or {@code frontierStartTime} when no files exist); + * lastRoundInSync is derived from the minimum of the record's lastSyncStateTimeInMs and that file + * frontier, so it never sits ahead of the last synced data. When lastSyncStateTimeInMs is 0 (no + * known sync point), it falls back to the file frontier so lastRoundInSync collapses onto + * lastRoundProcessed instead of rewinding to the epoch. + * @param haGroupStoreRecord the persisted HA group record supplying lastSyncStateTimeInMs + * @param frontierStartTime current time sampled at the start of initialization; the file frontier + * upper bound, and the sole basis when no files exist + */ + private void initLastRoundsFromLastSyncPoint(HAGroupStoreRecord haGroupStoreRecord, + long frontierStartTime) throws IOException { + long minimumTimestampFromFiles = frontierStartTime; + Optional minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles(); + Optional minTimestampFromNewFiles = getMinTimestampFromNewFiles(); + if (minTimestampFromInProgressFiles.isPresent()) { + LOG.info("Found minimum timestamp from IN PROGRESS files as {}", + minTimestampFromInProgressFiles.get()); + minimumTimestampFromFiles = + Math.min(minimumTimestampFromFiles, minTimestampFromInProgressFiles.get()); + } + if (minTimestampFromNewFiles.isPresent()) { + LOG.info("Found minimum timestamp from IN files as {}", minTimestampFromNewFiles.get()); + minimumTimestampFromFiles = + Math.min(minimumTimestampFromFiles, minTimestampFromNewFiles.get()); + } + this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager() + .getReplicationRoundFromEndTime(minimumTimestampFromFiles); + // A lastSyncStateTimeInMs of 0 means "no known sync point" (never synced, or a record that + // predates the field). Do NOT feed 0 into the min: getReplicationRoundFromEndTime(0) returns + // ReplicationRound(0, 0), which would rewind SYNCED_RECOVERY to the epoch and drive the + // consistency point to 0 (retain-everything). Fall back to the file frontier so lastRoundInSync + // collapses onto lastRoundProcessed (no rewind) when there is no known sync point. + long lastSyncStateTimeInMs = haGroupStoreRecord.getLastSyncStateTimeInMs(); + long lastSyncBasis = lastSyncStateTimeInMs > 0L + ? Math.min(lastSyncStateTimeInMs, minimumTimestampFromFiles) + : minimumTimestampFromFiles; + this.lastRoundInSync = replicationLogTracker.getReplicationShardDirectoryManager() + .getReplicationRoundFromEndTime(lastSyncBasis); + } + /** * Executes a replay operation with state-aware processing for HA replication scenarios. This * method extends the base replay() by handling three replication states: 1. SYNC: Normal 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 6e3b1cc85ec..f46638eeb76 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 @@ -350,6 +350,31 @@ public void testInitializeLastRoundProcessed_DegradedStateWithNoFiles() throws I false); } + @Test + public void testInitializeLastRoundProcessed_DegradedStateWithZeroLastSyncTime() + throws IOException { + long newFileTimestamp = 1704240060000L; + long lastSyncStateTime = 0L; // never synced / unset - must NOT rewind to the epoch + long currentTime = 1704240900000L; + long roundTimeMills = 60000L; // 1 minute + + // lastRoundProcessed uses the minimum of new files and current time (the file frontier). + long expectedEndTime = + (newFileTimestamp / TimeUnit.MINUTES.toMillis(1)) * TimeUnit.MINUTES.toMillis(1); + ReplicationRound expectedLastRoundProcessed = + new ReplicationRound(expectedEndTime - roundTimeMills, expectedEndTime); + + // The zero-lastSyncStateTimeInMs guard is shared by the DEGRADED_STANDBY and STANDBY_TO_ACTIVE + // init paths. On the DEGRADED path too, lastRoundInSync must collapse onto the file frontier + // (== lastRoundProcessed) instead of ReplicationRound(0, 0), which would rewind all the way to + // the epoch and drive the consistency point to 0 (retain-everything). + ReplicationRound expectedLastRoundInSync = expectedLastRoundProcessed; + + testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, + HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY, expectedLastRoundProcessed, + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED, false); + } + @Test public void testInitializeLastRoundProcessed_SyncStateWithInProgressFiles() throws IOException { long currentTime = 1704412800000L; @@ -461,7 +486,61 @@ public void testInitializeLastRoundProcessed_StandbyToActiveState() throws IOExc testInitializeLastRoundProcessedHelper(currentTime, null, null, null, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, - expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, true); + expectedLastRoundInSync, + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, true); + } + + @Test + public void testInitializeLastRoundProcessed_StandbyToActiveStateWithLastSyncStateAsMin() + throws IOException { + long newFileTimestamp = 1704240060000L; + long lastSyncStateTime = 1704240030000L; + long currentTime = 1704240900000L; + long roundTimeMills = 60000L; // 1 minute + + // lastRoundProcessed uses the minimum of new files and current time (the file frontier). + long expectedEndTime = + (newFileTimestamp / TimeUnit.MINUTES.toMillis(1)) * TimeUnit.MINUTES.toMillis(1); + ReplicationRound expectedLastRoundProcessed = + new ReplicationRound(expectedEndTime - roundTimeMills, expectedEndTime); + + // lastRoundInSync uses the minimum of lastSyncStateTime and file timestamps, so on restart into + // STANDBY_TO_ACTIVE it stays one round BEHIND lastRoundProcessed (not collapsed onto it) - this + // is the rewind target that guarantees zero RPO after a direct DEGRADED_STANDBY transition. + long expectedSyncEndTime = + (lastSyncStateTime / TimeUnit.MINUTES.toMillis(1)) * TimeUnit.MINUTES.toMillis(1); + ReplicationRound expectedLastRoundInSync = + new ReplicationRound(expectedSyncEndTime - roundTimeMills, expectedSyncEndTime); + + testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, + HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, + expectedLastRoundInSync, + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, true); + } + + @Test + public void testInitializeLastRoundProcessed_StandbyToActiveStateWithZeroLastSyncTime() + throws IOException { + long newFileTimestamp = 1704240060000L; + long lastSyncStateTime = 0L; // never synced / unset - must NOT rewind to the epoch + long currentTime = 1704240900000L; + long roundTimeMills = 60000L; // 1 minute + + // lastRoundProcessed uses the minimum of new files and current time (the file frontier). + long expectedEndTime = + (newFileTimestamp / TimeUnit.MINUTES.toMillis(1)) * TimeUnit.MINUTES.toMillis(1); + ReplicationRound expectedLastRoundProcessed = + new ReplicationRound(expectedEndTime - roundTimeMills, expectedEndTime); + + // With lastSyncStateTimeInMs == 0 (unset), lastRoundInSync must collapse onto the file frontier + // (== lastRoundProcessed) instead of ReplicationRound(0, 0), which would rewind SYNCED_RECOVERY + // all the way to the epoch and drive the consistency point to 0 (retain-everything). + ReplicationRound expectedLastRoundInSync = expectedLastRoundProcessed; + + testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, + HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, + expectedLastRoundInSync, + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, true); } /** @@ -1260,6 +1339,174 @@ public void testReplay_RuntimeListeners_DegradeAbortRecover() throws Exception { } } + /** + * PHOENIX-7920: drives the real LOCAL listeners through the DIRECT transition DEGRADED_STANDBY + * -> STANDBY_TO_ACTIVE (skipping STANDBY) via ZooKeeper, asserting that triggerFailoverListner + * promotes the replay state DEGRADED -> SYNCED_RECOVERY (so replay() will rewind) in addition + * to arming failoverPending. Before the fix the state stayed DEGRADED and failover was blocked + * forever. + */ + @Test + public void testReplay_RuntimeListeners_DirectDegradedToStandbyToActive() throws Exception { + PhoenixHAAdmin haAdmin = CLUSTERS.getHaAdmin1(); + // Base state: local STANDBY, written before init() so the discovery's client picks it up. + writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY); + + TestableReplicationLogTracker fileTracker = + createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri); + fileTracker.init(); + try { + // Null injected record: read the real effective record and subscribe the real LOCAL + // listeners to HAGroupStoreManager. + TestableReplicationLogDiscoveryReplay discovery = + new TestableReplicationLogDiscoveryReplay(fileTracker, null); + discovery.init(); + // Let the initial STANDBY load settle so it cannot race the degrade below. + Thread.sleep(5000L); + + // degrade: LOCAL -> DEGRADED_STANDBY must drive the degradedListener to DEGRADED. + writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY); + awaitCondition( + () -> discovery.getReplicationReplayState() + == ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED, + "degradedListener should drive replay state to DEGRADED"); + + // direct failover: LOCAL -> STANDBY_TO_ACTIVE (no STANDBY hop) must both arm failoverPending + // AND move DEGRADED -> SYNCED_RECOVERY so a rewind is scheduled. + writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE); + awaitCondition( + () -> discovery.getReplicationReplayState() + == ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, + "direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE must move replay state to SYNCED_RECOVERY"); + assertTrue("failoverPending must be set after STANDBY_TO_ACTIVE", + discovery.getFailoverPending()); + } finally { + fileTracker.close(); + haAdmin.getCurator().delete().quietly().forPath(toPath(haGroupName)); + } + } + + /** + * PHOENIX-7920: the healthy failover path STANDBY -> STANDBY_TO_ACTIVE (no prior degrade) must + * NOT change the replay state. init from STANDBY leaves state SYNC, and the trigger listener's + * compareAndSet(DEGRADED, SYNCED_RECOVERY) is a no-op from SYNC, arming only failoverPending. This + * guards against a naive unconditional set(SYNCED_RECOVERY) that would force a needless rewind on + * every healthy failover. + */ + @Test + public void testReplay_RuntimeListeners_HealthyStandbyToActiveKeepsSync() throws Exception { + PhoenixHAAdmin haAdmin = CLUSTERS.getHaAdmin1(); + writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY); + + TestableReplicationLogTracker fileTracker = + createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri); + fileTracker.init(); + try { + TestableReplicationLogDiscoveryReplay discovery = + new TestableReplicationLogDiscoveryReplay(fileTracker, null); + discovery.init(); + // Let the initial STANDBY load settle so init has resolved to SYNC before we proceed. + Thread.sleep(5000L); + assertEquals("init from STANDBY should leave replay state SYNC", + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); + + // healthy failover: LOCAL -> STANDBY_TO_ACTIVE from SYNC must arm failoverPending... + writeLocalRecord(haAdmin, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE); + awaitCondition(discovery::getFailoverPending, + "triggerFailoverListner should set failoverPending"); + // ...and must leave the state at SYNC (CAS from DEGRADED is a no-op here). Sleep briefly so a + // (wrong) unconditional set(SYNCED_RECOVERY) would have landed before we assert. + Thread.sleep(2000L); + assertEquals("STANDBY_TO_ACTIVE from SYNC must not change replay state", + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); + } finally { + fileTracker.close(); + haAdmin.getCurator().delete().quietly().forPath(toPath(haGroupName)); + } + } + + /** + * PHOENIX-7920 direct-path outcome: with a failover pending during degradation, once the rewind + * (SYNCED_RECOVERY) completes and replay returns to SYNC with everything caught up, failover is + * triggered exactly once. Complements testReplay_StateTransition_DegradedToAbortToRecover (the + * aborted case, where triggerFailover must NOT fire) and testReplay_StateTransition_ + * DegradedToSyncedRecovery (rewind with no pending failover, where it also must NOT fire). + */ + @Test + public void testReplay_DirectFailover_RewindsThenTriggersFailoverOnce() throws IOException { + TestableReplicationLogTracker fileTracker = + createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri); + + try { + long initialEndTime = 1704585600000L; // 2024-01-07 00:00:00 + long roundTimeMills = + fileTracker.getReplicationShardDirectoryManager().getReplicationRoundDurationSeconds() + * 1000L; + long bufferMillis = (long) (roundTimeMills * 0.15); + + // DEGRADED record (peer-blind degraded standby). + HAGroupStoreRecord mockRecord = + new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, haGroupName, + HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY, initialEndTime, + HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl, CLUSTERS.getMasterAddress1(), + CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L); + + // Allow processing up to 5 rounds. + long currentTime = initialEndTime + (5 * roundTimeMills) + bufferMillis; + EnvironmentEdge edge = () -> currentTime; + EnvironmentEdgeManager.injectEdge(edge); + + try { + TestableReplicationLogDiscoveryReplay discovery = + new TestableReplicationLogDiscoveryReplay(fileTracker, mockRecord); + discovery.init(); + + // Large rewind span: lastRoundProcessed is well ahead of the frozen lastRoundInSync. + ReplicationRound lastInSyncRound = + new ReplicationRound(initialEndTime - roundTimeMills, initialEndTime); + discovery.setLastRoundProcessed(new ReplicationRound(initialEndTime + roundTimeMills, + initialEndTime + (2 * roundTimeMills))); + discovery.setLastRoundInSync(lastInSyncRound); + discovery + .setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED); + // Failover requested during degradation, then the direct transition schedules the rewind + // (SYNCED_RECOVERY) after round 2. failoverPending is never cleared - it must survive. + discovery.setFailoverPending(true); + discovery.setStateChangeAfterRounds(2, + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY); + + discovery.replay(); + + // 2 rounds in DEGRADED, rewind to lastRoundInSync, then 5 rounds re-replayed in SYNC. + assertEquals("processRound should be called 7 times", 7, + discovery.getProcessRoundCallCount()); + + // Recovery completed: back to SYNC, both pointers caught up to the final round. + assertEquals("State should be SYNC after recovery", + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); + ReplicationRound expectedFinalRound = new ReplicationRound( + initialEndTime + (4 * roundTimeMills), initialEndTime + (5 * roundTimeMills)); + assertEquals("Last round processed should be the final round", expectedFinalRound, + discovery.getLastRoundProcessed()); + assertEquals("Last round in sync should match last round processed after recovery", + expectedFinalRound, discovery.getLastRoundInSync()); + + // The pending failover survived the rewind and fired exactly once, from SYNC. + assertEquals("Failover should be triggered exactly once after rewind completes", 1, + discovery.getTriggerFailoverCallCount()); + assertFalse("failoverPending should be cleared after failover is triggered", + discovery.getFailoverPending()); + } finally { + EnvironmentEdgeManager.reset(); + } + } finally { + fileTracker.close(); + } + } + /** Write (create, or version-checked update) the LOCAL HA record for this group on ZooKeeper. */ private void writeLocalRecord(PhoenixHAAdmin haAdmin, HAGroupStoreRecord.HAGroupState state) throws Exception { From f713e2408bf775424858f73587517ec0d8c80b4b Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 16 Jul 2026 17:01:24 +0530 Subject: [PATCH 2/4] Adding E2E test including HA Store changes --- .../CrossClusterReplicationTestUtil.java | 117 ++++++ .../replication/ReplicationLogGroupIT.java | 65 +-- .../ReplicationLogGroupTestAccess.java | 67 ++++ .../reader/StoreAndForwardFailoverIT.java | 373 ++++++++++++++++++ 4 files changed, 559 insertions(+), 63 deletions(-) create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java new file mode 100644 index 00000000000..0bd8b835594 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java @@ -0,0 +1,117 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.replication; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.ResultScanner; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.client.Table; +import org.apache.phoenix.util.TestUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Shared cross-cluster helpers for replication ITs (extracted from ReplicationLogGroupIT). */ +public final class CrossClusterReplicationTestUtil { + + private static final Logger LOG = + LoggerFactory.getLogger(CrossClusterReplicationTestUtil.class); + + private CrossClusterReplicationTestUtil() { + } + + /** Recursively collect every ".plog" file under {@code dir}; empty list if {@code dir} absent. */ + public static List findLogFiles(Path dir, FileSystem fs) throws IOException { + List files = new ArrayList<>(); + findLogFilesRecursive(dir, fs, files); + return files; + } + + private static void findLogFilesRecursive(Path dir, FileSystem fs, List files) + throws IOException { + if (!fs.exists(dir)) { + return; + } + for (FileStatus status : fs.listStatus(dir)) { + if (status.isDirectory()) { + findLogFilesRecursive(status.getPath(), fs, files); + } else if (status.getPath().getName().endsWith(".plog")) { + files.add(status.getPath()); + } + } + } + + /** + * Assert the given HBase table has cell-identical rows (all versions) on cluster 1 ({@code conf1}) + * and cluster 2 ({@code conf2}). Dumps both tables on the first mismatch before failing. + */ + public static void assertTablesEqualAcrossClusters(Configuration conf1, Configuration conf2, + String hbaseTableName) throws Exception { + TableName tn = TableName.valueOf(hbaseTableName); + try (Connection hconn1 = ConnectionFactory.createConnection(conf1); + Connection hconn2 = ConnectionFactory.createConnection(conf2); + Table table1 = hconn1.getTable(tn); Table table2 = hconn2.getTable(tn)) { + + Scan scan = new Scan(); + scan.readAllVersions(); + + try (ResultScanner scanner1 = table1.getScanner(scan); + ResultScanner scanner2 = table2.getScanner(scan)) { + int rowCount = 0; + while (true) { + Result r1 = scanner1.next(); + Result r2 = scanner2.next(); + if (r1 == null && r2 == null) { + break; + } + assertNotNull( + String.format("Table %s: cluster 2 has fewer rows at row %d", hbaseTableName, rowCount), + r2); + assertNotNull( + String.format("Table %s: cluster 1 has fewer rows at row %d", hbaseTableName, rowCount), + r1); + try { + Result.compareResults(r1, r2, true); + } catch (Exception e) { + LOG.error("Table {} row {} mismatch. Dumping both tables:", hbaseTableName, rowCount); + LOG.error("--- Cluster 1 ---"); + TestUtil.dumpTable(table1); + LOG.error("--- Cluster 2 ---"); + TestUtil.dumpTable(table2); + fail(String.format("Table %s row %d mismatch: %s", hbaseTableName, rowCount, + e.getMessage())); + } + rowCount++; + } + LOG.info("Table {} matches across clusters: {} rows verified", hbaseTableName, rowCount); + } + } + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java index c4dd5149b20..c95a4125051 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java @@ -32,27 +32,20 @@ import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; -import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.stream.Collectors; -import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.MiniHBaseCluster; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.TableName; -import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.Mutation; import org.apache.hadoop.hbase.client.RegionLocator; -import org.apache.hadoop.hbase.client.Result; -import org.apache.hadoop.hbase.client.ResultScanner; -import org.apache.hadoop.hbase.client.Scan; -import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.JVMClusterUtil; @@ -722,64 +715,10 @@ public void testSystemTables() throws Exception { } private List findLogFiles(Path dir, FileSystem fs) throws IOException { - List files = new ArrayList<>(); - findLogFilesRecursive(dir, fs, files); - return files; - } - - private void findLogFilesRecursive(Path dir, FileSystem fs, List files) throws IOException { - if (!fs.exists(dir)) { - return; - } - for (FileStatus status : fs.listStatus(dir)) { - if (status.isDirectory()) { - findLogFilesRecursive(status.getPath(), fs, files); - } else if (status.getPath().getName().endsWith(".plog")) { - files.add(status.getPath()); - } - } + return CrossClusterReplicationTestUtil.findLogFiles(dir, fs); } private void assertTablesEqualAcrossClusters(String hbaseTableName) throws Exception { - TableName tn = TableName.valueOf(hbaseTableName); - try ( - org.apache.hadoop.hbase.client.Connection hconn1 = ConnectionFactory.createConnection(conf1); - org.apache.hadoop.hbase.client.Connection hconn2 = ConnectionFactory.createConnection(conf2); - Table table1 = hconn1.getTable(tn); Table table2 = hconn2.getTable(tn)) { - - Scan scan = new Scan(); - scan.readAllVersions(); - - try (ResultScanner scanner1 = table1.getScanner(scan); - ResultScanner scanner2 = table2.getScanner(scan)) { - int rowCount = 0; - while (true) { - Result r1 = scanner1.next(); - Result r2 = scanner2.next(); - if (r1 == null && r2 == null) { - break; - } - assertNotNull( - String.format("Table %s: cluster 2 has fewer rows at row %d", hbaseTableName, rowCount), - r2); - assertNotNull( - String.format("Table %s: cluster 1 has fewer rows at row %d", hbaseTableName, rowCount), - r1); - try { - Result.compareResults(r1, r2, true); - } catch (Exception e) { - LOG.error("Table {} row {} mismatch. Dumping both tables:", hbaseTableName, rowCount); - LOG.error("--- Cluster 1 ---"); - TestUtil.dumpTable(table1); - LOG.error("--- Cluster 2 ---"); - TestUtil.dumpTable(table2); - fail(String.format("Table %s row %d mismatch: %s", hbaseTableName, rowCount, - e.getMessage())); - } - rowCount++; - } - LOG.info("Table {} matches across clusters: {} rows verified", hbaseTableName, rowCount); - } - } + CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1, conf2, hbaseTableName); } } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java new file mode 100644 index 00000000000..f72197a1e96 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +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 java.io.IOException; +import org.apache.hadoop.fs.Path; + +/** + * Integration-test-only access shim for the package-visible mode controls on + * {@link ReplicationLogGroup}. Lets an IT in another package flip a live writer SYNC -> + * STORE_AND_FORWARD deterministically, exercising the REAL {@link StoreAndForwardModeImpl} + * (local 'out' log + forwarder + periodic ACTIVE_NOT_IN_SYNC persistence) without having to + * injure a real peer HDFS to provoke the sync failure that triggers store-and-forward in + * production. Lives in the test source tree, so no production code is modified. + */ +public final class ReplicationLogGroupTestAccess { + + private ReplicationLogGroupTestAccess() { + } + + /** + * Flip a live writer from SYNC to STORE_AND_FORWARD and fence the swap through sync boundaries. + * The event handler reads the mode field in processPendingSyncs *after* completing a sync's + * future, so we issue two syncs: the first lets the handler observe the new mode and run + * StoreAndForwardModeImpl.onEnter (open the local 'out' log, start the forwarder); the second + * runs entirely under the store-and-forward mode impl, guaranteeing any subsequent append is + * buffered locally rather than written straight to the peer. + * @return true if the mode was SYNC and is now STORE_AND_FORWARD; false if it was not SYNC. + */ + public static boolean forceStoreAndForward(ReplicationLogGroup logGroup) throws IOException { + boolean swapped = logGroup.checkAndSetMode(SYNC, STORE_AND_FORWARD); + logGroup.sync(); + logGroup.sync(); + return swapped; + } + + /** True if the writer's current mode is STORE_AND_FORWARD. */ + public static boolean isStoreAndForward(ReplicationLogGroup logGroup) { + return logGroup.getMode() == STORE_AND_FORWARD; + } + + /** + * The peer (standby) 'in' directory this writer forwards to — the exact path a replay instance + * on the peer cluster must read. + */ + public static Path peerStandbyDir(ReplicationLogGroup logGroup) throws IOException { + return logGroup.getOrCreatePeerShardManager().getRootDirectoryPath(); + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java new file mode 100644 index 00000000000..9d487896d8d --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java @@ -0,0 +1,373 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.replication.reader; + +import static org.apache.phoenix.jdbc.HighAvailabilityGroup.PHOENIX_HA_GROUP_ATTR; +import static org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.getHighAvailibilityGroup; +import static org.apache.phoenix.query.BaseTest.generateUniqueName; +import static org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY; +import static org.apache.phoenix.replication.reader.ReplicationLogReplayService.PHOENIX_REPLICATION_REPLAY_ENABLED; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.Optional; +import java.util.Properties; +import java.util.function.BooleanSupplier; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.regionserver.HRegionServer; +import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest; +import org.apache.phoenix.jdbc.FailoverPhoenixConnection; +import org.apache.phoenix.jdbc.HABaseIT; +import org.apache.phoenix.jdbc.HAGroupStoreClient; +import org.apache.phoenix.jdbc.HAGroupStoreManager; +import org.apache.phoenix.jdbc.HAGroupStoreRecord; +import org.apache.phoenix.jdbc.HighAvailabilityGroup; +import org.apache.phoenix.jdbc.HighAvailabilityPolicy; +import org.apache.phoenix.jdbc.HighAvailabilityTestingUtility; +import org.apache.phoenix.replication.CrossClusterReplicationTestUtil; +import org.apache.phoenix.replication.ReplicationLogGroup; +import org.apache.phoenix.replication.ReplicationLogGroupTestAccess; +import org.apache.phoenix.replication.ReplicationLogTracker; +import org.apache.phoenix.replication.ReplicationShardDirectoryManager; +import org.apache.phoenix.replication.metrics.MetricsReplicationLogTrackerReplayImpl; +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.rules.TestName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * End-to-end zero-RPO test for a direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE failover in + * store-and-forward mode (PHOENIX-7920). Cluster1 runs the real writer; Cluster2 runs a + * manually-driven real ReplicationLogDiscoveryReplay (the auto-scheduler is disabled). + */ +@Category(NeedsOwnMiniClusterTest.class) +public class StoreAndForwardFailoverIT extends HABaseIT { + + private static final Logger LOG = LoggerFactory.getLogger(StoreAndForwardFailoverIT.class); + + private static final int A_START = 0; + private static final int A_COUNT = 50; + private static final int B_START = 50; + private static final int B_COUNT = 50; + + @Rule + public TestName name = new TestName(); + + private String haGroupName; + private Properties clientProps; + private HighAvailabilityGroup haGroup; + private ReplicationLogGroup logGroup; + private ReplicationLogDiscoveryReplay discovery; + private ReplicationLogTracker replayTracker; + private FileSystem standbyFs; + private Path standbyInDir; + private String tableName; + private long syncPointAfterA; + private int logFileCountAfterA; + + @BeforeClass + public static synchronized void doSetup() throws Exception { + // Short rounds so replay-round eligibility is a few seconds, not the 60s production default. + conf1.setInt(PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY, 2); + conf2.setInt(PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY, 2); + // Disable the auto replay scheduler so our manually-driven instance is the only replayer. + // (HABaseIT.doBaseSetup enables it; the writer/forwarder are gated on SYNCHRONOUS_REPLICATION, + // not on this flag, so they are unaffected.) + conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); + conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); + CLUSTERS.start(); + } + + @Before + public void beforeTest() throws Exception { + LOG.info("Starting test {}", name.getMethodName()); + haGroupName = name.getMethodName(); + clientProps = HighAvailabilityTestingUtility.getHATestProperties(); + clientProps.setProperty(PHOENIX_HA_GROUP_ATTR, haGroupName); + // Cluster1 = ACTIVE_IN_SYNC, Cluster2 = STANDBY, with hdfs urls populated on both records. + CLUSTERS.initClusterRole(haGroupName, HighAvailabilityPolicy.FAILOVER); + haGroup = getHighAvailibilityGroup(CLUSTERS.getJdbcHAUrl(), clientProps); + + // The live Cluster1 writer (RS0). Same singleton the RS write path uses, so flipping its mode + // affects real writes. Cluster1 is ACTIVE_IN_SYNC => this starts in SYNC mode. + HRegionServer rs = CLUSTERS.getHBaseCluster1().getHBaseCluster().getRegionServer(0); + logGroup = ReplicationLogGroup.get(conf1, rs.getServerName(), haGroupName); + + // Replicated table (REPLICATION_SCOPE=1 + LOCAL INDEX) on BOTH clusters. + tableName = "T_" + generateUniqueName(); + CLUSTERS.createTableOnClusterPair(haGroup, tableName); + + // Ensure Cluster2 sees its peer, so its effective HA record is STANDBY (not the peer-blind + // DEGRADED_STANDBY). This guarantees the replay initializes in SYNC (else-branch). + HAGroupStoreManager c2Manager = HAGroupStoreManager.getInstance(conf2); + awaitCondition(() -> { + try { + Optional eff = c2Manager.getEffectiveHAGroupStoreRecord(haGroupName); + return eff.isPresent() + && eff.get().getHAGroupState() == HAGroupStoreRecord.HAGroupState.STANDBY; + } catch (IOException e) { + return false; + } + }, 60000L, "cluster2 effective state should settle to STANDBY before constructing replay"); + + // Build the REAL replay over the exact standby 'in' directory the writer forwards to, while + // Cluster2 is still STANDBY, so init() subscribes its LOCAL listeners BEFORE any degrade / + // failover write and starts in SYNC. + standbyInDir = ReplicationLogGroupTestAccess.peerStandbyDir(logGroup); + standbyFs = standbyInDir.getFileSystem(conf2); + ReplicationShardDirectoryManager shardMgr = + new ReplicationShardDirectoryManager(conf2, standbyFs, standbyInDir); + replayTracker = new ReplicationLogTracker(conf2, haGroupName, shardMgr, + new MetricsReplicationLogTrackerReplayImpl(haGroupName)); + replayTracker.init(); + discovery = new ReplicationLogDiscoveryReplay(replayTracker); + discovery.init(); + } + + @After + public void afterTest() throws Exception { + LOG.info("Cleaning up test {}", name.getMethodName()); + if (replayTracker != null) { + replayTracker.close(); + } + if (logGroup != null) { + logGroup.close(); + } + } + + @Test + public void testDirectFailoverInStoreAndForwardModeIsZeroRPO() throws Exception { + // Stage 0: the replay initialized in SYNC while Cluster2 is STANDBY. + assertEquals("replay must initialize in SYNC while cluster2 is STANDBY", + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); + // Stage 1: SYNC batch A replicates to cluster2 and advances the sync point. + stageWriteAndReplayBatchA(); + // Stage 2: enter store-and-forward, write batch B, await it forwarded to cluster2 'in'. + stageEnterStoreAndForwardAndForwardBatchB(); + // Stage 3: degrade, direct failover, drive replay to promotion, assert zero RPO. + stageDegradeDirectFailoverAndAssertZeroRPO(); + } + + /** Poll until {@code condition} holds or {@code timeoutMs} elapses, then assert it. */ + private static void awaitCondition(BooleanSupplier condition, long timeoutMs, String message) + throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMs; + while (!condition.getAsBoolean() && System.currentTimeMillis() < deadline) { + Thread.sleep(250L); + } + assertTrue(message, condition.getAsBoolean()); + } + + /** + * Repeatedly call {@code discovery.replay()} (short real-clock rounds) until {@code done} holds or + * {@code timeoutMs} elapses, then assert {@code done}. This is how the test advances the manually + * driven replayer in the absence of the auto-scheduler. + */ + private void driveReplayUntil(BooleanSupplier done, long timeoutMs, String message) + throws Exception { + long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + discovery.replay(); + if (done.getAsBoolean()) { + return; + } + Thread.sleep(500L); + } + discovery.replay(); + assertTrue(message, done.getAsBoolean()); + } + + /** Upsert {@code count} rows [startId, startId+count) into the replicated table via the HA URL. */ + private void upsertRows(int startId, int count) throws SQLException { + try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection) DriverManager + .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) { + PreparedStatement stmt = + conn.prepareStatement("UPSERT INTO " + tableName + " VALUES (?, ?)"); + for (int i = 0; i < count; i++) { + stmt.setInt(1, startId + i); + stmt.setInt(2, startId + i); + stmt.executeUpdate(); + } + conn.commit(); + } + } + + /** + * Stage 1: write batch A on Cluster1 in SYNC mode (.plog goes straight to Cluster2's 'in' dir), + * then drive replay until every A row is present on Cluster2 and lastRoundInSync has advanced to + * cover A. Records the sync point so Task 5 can assert it later advances to cover B. + */ + private void stageWriteAndReplayBatchA() throws Exception { + upsertRows(A_START, A_COUNT); + + // Drive replay until A has been applied to Cluster2 and the sync point advanced past epoch. + driveReplayUntil(() -> { + try { + CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1, conf2, tableName); + return discovery.getLastRoundInSync() != null + && discovery.getLastRoundInSync().getEndTime() > 0L; + } catch (AssertionError notYet) { + return false; + } catch (Exception e) { + return false; + } + }, 90000L, "batch A must replicate to cluster2 and advance lastRoundInSync"); + + syncPointAfterA = discovery.getLastRoundInSync().getEndTime(); + assertTrue("lastRoundInSync must be set after batch A", syncPointAfterA > 0L); + LOG.info("Stage 1 complete: batch A in sync, syncPointAfterA={}", syncPointAfterA); + } + + /** + * Stage 2: flip the live Cluster1 writer SYNC -> STORE_AND_FORWARD (real StoreAndForwardModeImpl + * onEnter: local 'out' log + forwarder + periodic ACTIVE_NOT_IN_SYNC persistence on Cluster1), + * write batch B (buffered to Cluster1 'out'), and wait for the real forwarder to copy B's .plog + * files into Cluster2's 'in' directory. Does NOT drive replay here, so B stays unreplayed. + */ + private void stageEnterStoreAndForwardAndForwardBatchB() throws Exception { + logFileCountAfterA = CrossClusterReplicationTestUtil.findLogFiles(standbyInDir, standbyFs).size(); + + boolean swapped = ReplicationLogGroupTestAccess.forceStoreAndForward(logGroup); + assertTrue("writer must have been in SYNC and flipped to STORE_AND_FORWARD", swapped); + assertTrue("writer must now report STORE_AND_FORWARD", + ReplicationLogGroupTestAccess.isStoreAndForward(logGroup)); + + upsertRows(B_START, B_COUNT); + + // The forwarder copies out->in on its own executor; poll (do not race) for B's new files. + awaitCondition(() -> { + try { + return CrossClusterReplicationTestUtil.findLogFiles(standbyInDir, standbyFs).size() + > logFileCountAfterA; + } catch (IOException e) { + return false; + } + }, 60000L, "forwarder must copy batch B's .plog files into cluster2 'in'"); + LOG.info("Stage 2 complete: store-and-forward active, batch B forwarded to cluster2 'in'"); + } + + /** + * Drive Cluster2's LOCAL HA record to {@code target} through the same cached client the real + * triggerFailover uses. setHAGroupStatusIfNeeded returns a positive dwell-time when the + * transition is throttled; retry until it applies (returns 0) or the deadline elapses. + * + *

Cluster2 is peer-aware: when Cluster1 persists {@code ACTIVE_NOT_IN_SYNC} on entering + * store-and-forward, Cluster2 auto-reacts and drives its own LOCAL record to + * {@code DEGRADED_STANDBY}. That reaction may still be in flight when this helper runs, so we + * first poll a bounded window for the record to settle at {@code target} and treat "already at + * target" as a no-op — otherwise a redundant transition (e.g. DEGRADED_STANDBY -> + * DEGRADED_STANDBY) would throw InvalidClusterRoleTransitionException on a slow box. + */ + private void transitionCluster2(HAGroupStoreRecord.HAGroupState target) throws Exception { + // Let any in-flight peer-aware reaction settle so the no-op check below fires deterministically. + long settleDeadline = System.currentTimeMillis() + 10000L; + while (System.currentTimeMillis() < settleDeadline && !cluster2StateIs(target)) { + Thread.sleep(500L); + } + // Check if already at target - no-op if so (avoid redundant transition exception). + if (cluster2StateIs(target)) { + return; + } + HAGroupStoreClient client = HAGroupStoreClient.getInstance(conf2, haGroupName); + long deadline = System.currentTimeMillis() + 60000L; + long lastWait = -1L; + while (System.currentTimeMillis() < deadline) { + lastWait = client.setHAGroupStatusIfNeeded(target); + if (lastWait == 0L) { + return; + } + Thread.sleep(Math.min(lastWait, 2000L)); + } + throw new AssertionError("cluster2 transition to " + target + + " was still throttled at deadline (lastWait=" + lastWait + ")"); + } + + /** True if Cluster2's persisted HA record is currently {@code state}. */ + private boolean cluster2StateIs(HAGroupStoreRecord.HAGroupState state) { + try { + Optional rec = + HAGroupStoreManager.getInstance(conf2).getHAGroupStoreRecord(haGroupName); + return rec.isPresent() && rec.get().getHAGroupState() == state; + } catch (IOException e) { + return false; + } + } + + /** + * Stage 3: degrade Cluster2 (freezes lastRoundInSync at A), perform the real direct + * DEGRADED_STANDBY -> STANDBY_TO_ACTIVE, then drive replay until Cluster2 promotes to + * ACTIVE_IN_SYNC. Assert zero RPO: every A and B row is present cell-for-cell on Cluster2, the + * replay state is back to SYNC, and lastRoundInSync advanced past its post-A value to cover B. + * + *

Regression guard: without PHOENIX-7920's triggerFailoverListner CAS(DEGRADED, + * SYNCED_RECOVERY), the direct transition leaves the replay state at DEGRADED forever, + * shouldTriggerFailover() never passes, Cluster2 never reaches ACTIVE_IN_SYNC, and the promotion + * poll below times out (red). + */ + private void stageDegradeDirectFailoverAndAssertZeroRPO() throws Exception { + // Degrade: LOCAL -> DEGRADED_STANDBY drives the degradedListener to DEGRADED and freezes + // lastRoundInSync at the batch-A sync point (B is forwarded but not yet replayed). + transitionCluster2(HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY); + awaitCondition(() -> discovery.getReplicationReplayState() + == ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED, + 30000L, "degradedListener should drive replay state to DEGRADED"); + assertEquals("lastRoundInSync must stay frozen at the batch-A sync point during DEGRADED", + syncPointAfterA, discovery.getLastRoundInSync().getEndTime()); + + // Direct failover: LOCAL -> STANDBY_TO_ACTIVE (no STANDBY hop). triggerFailoverListner must + // CAS DEGRADED -> SYNCED_RECOVERY and arm failoverPending. + transitionCluster2(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE); + awaitCondition(() -> discovery.getReplicationReplayState() + == ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, + 30000L, "direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE must move replay to SYNCED_RECOVERY"); + assertTrue("failoverPending must be armed after STANDBY_TO_ACTIVE", + discovery.getFailoverPending()); + + // Make sure the client cache reflects STANDBY_TO_ACTIVE before triggerFailover reads it. + awaitCondition(() -> cluster2StateIs(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE), + 30000L, "cluster2 record should reflect STANDBY_TO_ACTIVE before driving failover"); + + logGroup.close(); + + // Drive replay: SYNCED_RECOVERY rewinds to lastRoundInSync (A), re-replays B in SYNC. With no + // new files arriving, the replay drains, shouldTriggerFailover() passes, and triggerFailover() + // sets ACTIVE_IN_SYNC on Cluster2. + driveReplayUntil(() -> cluster2StateIs(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC), + 120000L, "cluster2 must promote to ACTIVE_IN_SYNC after the direct failover"); + + // Zero-RPO assertions. + CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1, conf2, tableName); + assertEquals("replay state must be SYNC after promotion", + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); + assertTrue("lastRoundInSync must advance past the batch-A sync point to cover batch B", + discovery.getLastRoundInSync().getEndTime() > syncPointAfterA); + LOG.info("Stage 3 complete: cluster2 promoted to ACTIVE_IN_SYNC with zero RPO"); + } +} From 4348d69759750aaca591ff23e89a4687d2f21181 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Fri, 17 Jul 2026 13:40:36 +0530 Subject: [PATCH 3/4] PHOENIX-7920 Apply spotless formatting and inline findLogFiles test helper --- .../replication/ReplicationLogDiscovery.java | 5 +- .../reader/ReplicationLogDiscoveryReplay.java | 7 +- .../CrossClusterReplicationTestUtil.java | 8 +- .../replication/ReplicationLogGroupIT.java | 8 +- .../ReplicationLogGroupTestAccess.java | 12 +-- .../ReplicationLogDiscoveryReplayTestIT.java | 21 ++-- .../reader/StoreAndForwardFailoverIT.java | 98 ++++++++++--------- 7 files changed, 81 insertions(+), 78 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 83288cde44d..638d95280f5 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 @@ -400,8 +400,9 @@ protected void initializeLastRoundProcessed() throws IOException { * getReplicationRoundFromEndTime(), which rounds down to the nearest round boundary to ensure we * start from a complete round. * @param fallbackCurrentTime the timestamp to start from when no files exist; sampled by the - * caller at the start of initialization rather than re-read here, so a slow init (or a - * state transition during init) cannot push the starting round forward + * caller at the start of initialization rather than re-read here, so a + * slow init (or a state transition during init) cannot push the + * starting round forward * @throws IOException if there's an error reading file timestamps */ protected void initializeLastRoundProcessed(long fallbackCurrentTime) throws IOException { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java index b19cc988dd4..8cf469b7032 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java @@ -243,7 +243,8 @@ protected void initializeLastRoundProcessed() throws IOException { } else if (HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) { // Restarted while already in STANDBY_TO_ACTIVE (e.g. an RS bounce after a direct // DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). No listener fires on a fresh process, so - // initialize as if recovering: lastRoundInSync at the last good sync point, lastRoundProcessed + // initialize as if recovering: lastRoundInSync at the last good sync point, + // lastRoundProcessed // from the file frontier, and state SYNCED_RECOVERY so the first replay() rewinds before // shouldTriggerFailover() can promote. Conservative by design: if we actually came from a // healthy STANDBY, lastSyncStateTimeInMs ~= the file frontier and the rewind is a no-op. @@ -278,8 +279,8 @@ protected void initializeLastRoundProcessed() throws IOException { * known sync point), it falls back to the file frontier so lastRoundInSync collapses onto * lastRoundProcessed instead of rewinding to the epoch. * @param haGroupStoreRecord the persisted HA group record supplying lastSyncStateTimeInMs - * @param frontierStartTime current time sampled at the start of initialization; the file frontier - * upper bound, and the sole basis when no files exist + * @param frontierStartTime current time sampled at the start of initialization; the file + * frontier upper bound, and the sole basis when no files exist */ private void initLastRoundsFromLastSyncPoint(HAGroupStoreRecord haGroupStoreRecord, long frontierStartTime) throws IOException { diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java index 0bd8b835594..127f793415f 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/CrossClusterReplicationTestUtil.java @@ -41,8 +41,7 @@ /** Shared cross-cluster helpers for replication ITs (extracted from ReplicationLogGroupIT). */ public final class CrossClusterReplicationTestUtil { - private static final Logger LOG = - LoggerFactory.getLogger(CrossClusterReplicationTestUtil.class); + private static final Logger LOG = LoggerFactory.getLogger(CrossClusterReplicationTestUtil.class); private CrossClusterReplicationTestUtil() { } @@ -69,8 +68,9 @@ private static void findLogFilesRecursive(Path dir, FileSystem fs, List fi } /** - * Assert the given HBase table has cell-identical rows (all versions) on cluster 1 ({@code conf1}) - * and cluster 2 ({@code conf2}). Dumps both tables on the first mismatch before failing. + * Assert the given HBase table has cell-identical rows (all versions) on cluster 1 + * ({@code conf1}) and cluster 2 ({@code conf2}). Dumps both tables on the first mismatch before + * failing. */ public static void assertTablesEqualAcrossClusters(Configuration conf1, Configuration conf2, String hbaseTableName) throws Exception { diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java index c95a4125051..335cdee1adc 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java @@ -24,9 +24,7 @@ import static org.apache.phoenix.query.BaseTest.generateUniqueName; import static org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; import java.io.IOException; import java.sql.Connection; @@ -237,7 +235,7 @@ private void replayAndVerifyAcrossClusters(List ddlStatements, String... // Replay replication log on cluster 2 FileSystem fs = standByLogDir.getFileSystem(conf2); - List logFiles = findLogFiles(standByLogDir, fs); + List logFiles = CrossClusterReplicationTestUtil.findLogFiles(standByLogDir, fs); assertTrue("Should have at least one log file", !logFiles.isEmpty()); ReplicationLogProcessor processor = ReplicationLogProcessor.get(conf2, haGroupName); try { @@ -714,10 +712,6 @@ public void testSystemTables() throws Exception { assertTrue(getCountForTable(systemTables, SYSTEM_CATALOG_NAME) > 0); } - private List findLogFiles(Path dir, FileSystem fs) throws IOException { - return CrossClusterReplicationTestUtil.findLogFiles(dir, fs); - } - private void assertTablesEqualAcrossClusters(String hbaseTableName) throws Exception { CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1, conf2, hbaseTableName); } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java index f72197a1e96..1b363c8b0ee 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupTestAccess.java @@ -26,10 +26,10 @@ /** * Integration-test-only access shim for the package-visible mode controls on * {@link ReplicationLogGroup}. Lets an IT in another package flip a live writer SYNC -> - * STORE_AND_FORWARD deterministically, exercising the REAL {@link StoreAndForwardModeImpl} - * (local 'out' log + forwarder + periodic ACTIVE_NOT_IN_SYNC persistence) without having to - * injure a real peer HDFS to provoke the sync failure that triggers store-and-forward in - * production. Lives in the test source tree, so no production code is modified. + * STORE_AND_FORWARD deterministically, exercising the REAL {@link StoreAndForwardModeImpl} (local + * 'out' log + forwarder + periodic ACTIVE_NOT_IN_SYNC persistence) without having to injure a real + * peer HDFS to provoke the sync failure that triggers store-and-forward in production. Lives in the + * test source tree, so no production code is modified. */ public final class ReplicationLogGroupTestAccess { @@ -58,8 +58,8 @@ public static boolean isStoreAndForward(ReplicationLogGroup logGroup) { } /** - * The peer (standby) 'in' directory this writer forwards to — the exact path a replay instance - * on the peer cluster must read. + * The peer (standby) 'in' directory this writer forwards to — the exact path a replay instance on + * the peer cluster must read. */ public static Path peerStandbyDir(ReplicationLogGroup logGroup) throws IOException { return logGroup.getOrCreatePeerShardManager().getRootDirectoryPath(); 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 f46638eeb76..9eef0cd74cd 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 @@ -372,7 +372,8 @@ public void testInitializeLastRoundProcessed_DegradedStateWithZeroLastSyncTime() testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY, expectedLastRoundProcessed, - expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED, false); + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED, + false); } @Test @@ -486,8 +487,8 @@ public void testInitializeLastRoundProcessed_StandbyToActiveState() throws IOExc testInitializeLastRoundProcessedHelper(currentTime, null, null, null, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, - expectedLastRoundInSync, - ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, true); + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, + true); } @Test @@ -514,8 +515,8 @@ public void testInitializeLastRoundProcessed_StandbyToActiveStateWithLastSyncSta testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, - expectedLastRoundInSync, - ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, true); + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, + true); } @Test @@ -539,8 +540,8 @@ public void testInitializeLastRoundProcessed_StandbyToActiveStateWithZeroLastSyn testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, - expectedLastRoundInSync, - ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, true); + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, + true); } /** @@ -1389,9 +1390,9 @@ public void testReplay_RuntimeListeners_DirectDegradedToStandbyToActive() throws /** * PHOENIX-7920: the healthy failover path STANDBY -> STANDBY_TO_ACTIVE (no prior degrade) must * NOT change the replay state. init from STANDBY leaves state SYNC, and the trigger listener's - * compareAndSet(DEGRADED, SYNCED_RECOVERY) is a no-op from SYNC, arming only failoverPending. This - * guards against a naive unconditional set(SYNCED_RECOVERY) that would force a needless rewind on - * every healthy failover. + * compareAndSet(DEGRADED, SYNCED_RECOVERY) is a no-op from SYNC, arming only failoverPending. + * This guards against a naive unconditional set(SYNCED_RECOVERY) that would force a needless + * rewind on every healthy failover. */ @Test public void testReplay_RuntimeListeners_HealthyStandbyToActiveKeepsSync() throws Exception { diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java index 9d487896d8d..d3d5690cd8c 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java @@ -129,7 +129,7 @@ public void beforeTest() throws Exception { try { Optional eff = c2Manager.getEffectiveHAGroupStoreRecord(haGroupName); return eff.isPresent() - && eff.get().getHAGroupState() == HAGroupStoreRecord.HAGroupState.STANDBY; + && eff.get().getHAGroupState() == HAGroupStoreRecord.HAGroupState.STANDBY; } catch (IOException e) { return false; } @@ -141,9 +141,9 @@ public void beforeTest() throws Exception { standbyInDir = ReplicationLogGroupTestAccess.peerStandbyDir(logGroup); standbyFs = standbyInDir.getFileSystem(conf2); ReplicationShardDirectoryManager shardMgr = - new ReplicationShardDirectoryManager(conf2, standbyFs, standbyInDir); + new ReplicationShardDirectoryManager(conf2, standbyFs, standbyInDir); replayTracker = new ReplicationLogTracker(conf2, haGroupName, shardMgr, - new MetricsReplicationLogTrackerReplayImpl(haGroupName)); + new MetricsReplicationLogTrackerReplayImpl(haGroupName)); replayTracker.init(); discovery = new ReplicationLogDiscoveryReplay(replayTracker); discovery.init(); @@ -164,8 +164,8 @@ public void afterTest() throws Exception { public void testDirectFailoverInStoreAndForwardModeIsZeroRPO() throws Exception { // Stage 0: the replay initialized in SYNC while Cluster2 is STANDBY. assertEquals("replay must initialize in SYNC while cluster2 is STANDBY", - ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, - discovery.getReplicationReplayState()); + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); // Stage 1: SYNC batch A replicates to cluster2 and advances the sync point. stageWriteAndReplayBatchA(); // Stage 2: enter store-and-forward, write batch B, await it forwarded to cluster2 'in'. @@ -176,7 +176,7 @@ public void testDirectFailoverInStoreAndForwardModeIsZeroRPO() throws Exception /** Poll until {@code condition} holds or {@code timeoutMs} elapses, then assert it. */ private static void awaitCondition(BooleanSupplier condition, long timeoutMs, String message) - throws InterruptedException { + throws InterruptedException { long deadline = System.currentTimeMillis() + timeoutMs; while (!condition.getAsBoolean() && System.currentTimeMillis() < deadline) { Thread.sleep(250L); @@ -185,12 +185,12 @@ private static void awaitCondition(BooleanSupplier condition, long timeoutMs, St } /** - * Repeatedly call {@code discovery.replay()} (short real-clock rounds) until {@code done} holds or - * {@code timeoutMs} elapses, then assert {@code done}. This is how the test advances the manually - * driven replayer in the absence of the auto-scheduler. + * Repeatedly call {@code discovery.replay()} (short real-clock rounds) until {@code done} holds + * or {@code timeoutMs} elapses, then assert {@code done}. This is how the test advances the + * manually driven replayer in the absence of the auto-scheduler. */ private void driveReplayUntil(BooleanSupplier done, long timeoutMs, String message) - throws Exception { + throws Exception { long deadline = System.currentTimeMillis() + timeoutMs; while (System.currentTimeMillis() < deadline) { discovery.replay(); @@ -203,12 +203,13 @@ private void driveReplayUntil(BooleanSupplier done, long timeoutMs, String messa assertTrue(message, done.getAsBoolean()); } - /** Upsert {@code count} rows [startId, startId+count) into the replicated table via the HA URL. */ + /** + * Upsert {@code count} rows [startId, startId+count) into the replicated table via the HA URL. + */ private void upsertRows(int startId, int count) throws SQLException { try (FailoverPhoenixConnection conn = (FailoverPhoenixConnection) DriverManager - .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) { - PreparedStatement stmt = - conn.prepareStatement("UPSERT INTO " + tableName + " VALUES (?, ?)"); + .getConnection(CLUSTERS.getJdbcHAUrl(), clientProps)) { + PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " VALUES (?, ?)"); for (int i = 0; i < count; i++) { stmt.setInt(1, startId + i); stmt.setInt(2, startId + i); @@ -231,7 +232,7 @@ private void stageWriteAndReplayBatchA() throws Exception { try { CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1, conf2, tableName); return discovery.getLastRoundInSync() != null - && discovery.getLastRoundInSync().getEndTime() > 0L; + && discovery.getLastRoundInSync().getEndTime() > 0L; } catch (AssertionError notYet) { return false; } catch (Exception e) { @@ -245,18 +246,20 @@ private void stageWriteAndReplayBatchA() throws Exception { } /** - * Stage 2: flip the live Cluster1 writer SYNC -> STORE_AND_FORWARD (real StoreAndForwardModeImpl - * onEnter: local 'out' log + forwarder + periodic ACTIVE_NOT_IN_SYNC persistence on Cluster1), - * write batch B (buffered to Cluster1 'out'), and wait for the real forwarder to copy B's .plog - * files into Cluster2's 'in' directory. Does NOT drive replay here, so B stays unreplayed. + * Stage 2: flip the live Cluster1 writer SYNC -> STORE_AND_FORWARD (real + * StoreAndForwardModeImpl onEnter: local 'out' log + forwarder + periodic ACTIVE_NOT_IN_SYNC + * persistence on Cluster1), write batch B (buffered to Cluster1 'out'), and wait for the real + * forwarder to copy B's .plog files into Cluster2's 'in' directory. Does NOT drive replay here, + * so B stays unreplayed. */ private void stageEnterStoreAndForwardAndForwardBatchB() throws Exception { - logFileCountAfterA = CrossClusterReplicationTestUtil.findLogFiles(standbyInDir, standbyFs).size(); + logFileCountAfterA = + CrossClusterReplicationTestUtil.findLogFiles(standbyInDir, standbyFs).size(); boolean swapped = ReplicationLogGroupTestAccess.forceStoreAndForward(logGroup); assertTrue("writer must have been in SYNC and flipped to STORE_AND_FORWARD", swapped); assertTrue("writer must now report STORE_AND_FORWARD", - ReplicationLogGroupTestAccess.isStoreAndForward(logGroup)); + ReplicationLogGroupTestAccess.isStoreAndForward(logGroup)); upsertRows(B_START, B_COUNT); @@ -276,8 +279,8 @@ private void stageEnterStoreAndForwardAndForwardBatchB() throws Exception { * Drive Cluster2's LOCAL HA record to {@code target} through the same cached client the real * triggerFailover uses. setHAGroupStatusIfNeeded returns a positive dwell-time when the * transition is throttled; retry until it applies (returns 0) or the deadline elapses. - * - *

Cluster2 is peer-aware: when Cluster1 persists {@code ACTIVE_NOT_IN_SYNC} on entering + *

+ * Cluster2 is peer-aware: when Cluster1 persists {@code ACTIVE_NOT_IN_SYNC} on entering * store-and-forward, Cluster2 auto-reacts and drives its own LOCAL record to * {@code DEGRADED_STANDBY}. That reaction may still be in flight when this helper runs, so we * first poll a bounded window for the record to settle at {@code target} and treat "already at @@ -285,7 +288,8 @@ private void stageEnterStoreAndForwardAndForwardBatchB() throws Exception { * DEGRADED_STANDBY) would throw InvalidClusterRoleTransitionException on a slow box. */ private void transitionCluster2(HAGroupStoreRecord.HAGroupState target) throws Exception { - // Let any in-flight peer-aware reaction settle so the no-op check below fires deterministically. + // Let any in-flight peer-aware reaction settle so the no-op check below fires + // deterministically. long settleDeadline = System.currentTimeMillis() + 10000L; while (System.currentTimeMillis() < settleDeadline && !cluster2StateIs(target)) { Thread.sleep(500L); @@ -305,14 +309,14 @@ private void transitionCluster2(HAGroupStoreRecord.HAGroupState target) throws E Thread.sleep(Math.min(lastWait, 2000L)); } throw new AssertionError("cluster2 transition to " + target - + " was still throttled at deadline (lastWait=" + lastWait + ")"); + + " was still throttled at deadline (lastWait=" + lastWait + ")"); } /** True if Cluster2's persisted HA record is currently {@code state}. */ private boolean cluster2StateIs(HAGroupStoreRecord.HAGroupState state) { try { Optional rec = - HAGroupStoreManager.getInstance(conf2).getHAGroupStoreRecord(haGroupName); + HAGroupStoreManager.getInstance(conf2).getHAGroupStoreRecord(haGroupName); return rec.isPresent() && rec.get().getHAGroupState() == state; } catch (IOException e) { return false; @@ -324,50 +328,52 @@ private boolean cluster2StateIs(HAGroupStoreRecord.HAGroupState state) { * DEGRADED_STANDBY -> STANDBY_TO_ACTIVE, then drive replay until Cluster2 promotes to * ACTIVE_IN_SYNC. Assert zero RPO: every A and B row is present cell-for-cell on Cluster2, the * replay state is back to SYNC, and lastRoundInSync advanced past its post-A value to cover B. - * - *

Regression guard: without PHOENIX-7920's triggerFailoverListner CAS(DEGRADED, - * SYNCED_RECOVERY), the direct transition leaves the replay state at DEGRADED forever, - * shouldTriggerFailover() never passes, Cluster2 never reaches ACTIVE_IN_SYNC, and the promotion - * poll below times out (red). + *

+ * Regression guard: without PHOENIX-7920's triggerFailoverListner CAS(DEGRADED, SYNCED_RECOVERY), + * the direct transition leaves the replay state at DEGRADED forever, shouldTriggerFailover() + * never passes, Cluster2 never reaches ACTIVE_IN_SYNC, and the promotion poll below times out + * (red). */ private void stageDegradeDirectFailoverAndAssertZeroRPO() throws Exception { // Degrade: LOCAL -> DEGRADED_STANDBY drives the degradedListener to DEGRADED and freezes // lastRoundInSync at the batch-A sync point (B is forwarded but not yet replayed). transitionCluster2(HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY); - awaitCondition(() -> discovery.getReplicationReplayState() - == ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED, - 30000L, "degradedListener should drive replay state to DEGRADED"); + awaitCondition( + () -> discovery.getReplicationReplayState() + == ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED, + 30000L, "degradedListener should drive replay state to DEGRADED"); assertEquals("lastRoundInSync must stay frozen at the batch-A sync point during DEGRADED", - syncPointAfterA, discovery.getLastRoundInSync().getEndTime()); + syncPointAfterA, discovery.getLastRoundInSync().getEndTime()); // Direct failover: LOCAL -> STANDBY_TO_ACTIVE (no STANDBY hop). triggerFailoverListner must // CAS DEGRADED -> SYNCED_RECOVERY and arm failoverPending. transitionCluster2(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE); - awaitCondition(() -> discovery.getReplicationReplayState() - == ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, - 30000L, "direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE must move replay to SYNCED_RECOVERY"); + awaitCondition( + () -> discovery.getReplicationReplayState() + == ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, + 30000L, "direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE must move replay to SYNCED_RECOVERY"); assertTrue("failoverPending must be armed after STANDBY_TO_ACTIVE", - discovery.getFailoverPending()); + discovery.getFailoverPending()); // Make sure the client cache reflects STANDBY_TO_ACTIVE before triggerFailover reads it. - awaitCondition(() -> cluster2StateIs(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE), - 30000L, "cluster2 record should reflect STANDBY_TO_ACTIVE before driving failover"); + awaitCondition(() -> cluster2StateIs(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE), 30000L, + "cluster2 record should reflect STANDBY_TO_ACTIVE before driving failover"); logGroup.close(); // Drive replay: SYNCED_RECOVERY rewinds to lastRoundInSync (A), re-replays B in SYNC. With no // new files arriving, the replay drains, shouldTriggerFailover() passes, and triggerFailover() // sets ACTIVE_IN_SYNC on Cluster2. - driveReplayUntil(() -> cluster2StateIs(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC), - 120000L, "cluster2 must promote to ACTIVE_IN_SYNC after the direct failover"); + driveReplayUntil(() -> cluster2StateIs(HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC), 120000L, + "cluster2 must promote to ACTIVE_IN_SYNC after the direct failover"); // Zero-RPO assertions. CrossClusterReplicationTestUtil.assertTablesEqualAcrossClusters(conf1, conf2, tableName); assertEquals("replay state must be SYNC after promotion", - ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, - discovery.getReplicationReplayState()); + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); assertTrue("lastRoundInSync must advance past the batch-A sync point to cover batch B", - discovery.getLastRoundInSync().getEndTime() > syncPointAfterA); + discovery.getLastRoundInSync().getEndTime() > syncPointAfterA); LOG.info("Stage 3 complete: cluster2 promoted to ACTIVE_IN_SYNC with zero RPO"); } } From 05754ecb552cb86493764ff8b27f5a4fbc50743d Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Sat, 18 Jul 2026 18:21:02 +0530 Subject: [PATCH 4/4] PHOENIX-7920 SAF failover: restart into STANDBY_TO_ACTIVE inits to SYNC when nothing to rewind On a restart while already STANDBY_TO_ACTIVE (e.g. an RS bounce after a direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE), enter SYNCED_RECOVERY only when a rewind is actually pending (lastRoundInSync behind lastRoundProcessed). When the rounds are equal there is nothing to rewind, so initialize directly to SYNC and let the first replay() promote immediately, instead of waiting ~one round window for the SYNCED_RECOVERY -> SYNC CAS that only fires inside the round-processing loop. Arm failoverPending with a plain set and emit distinct restart-init log lines for the two sub-cases. Add regression guard testReplay_RestartIntoStandbyToActive_NothingToReplay_PromotesOnFirstTick (promotes on the first replay tick) and update the existing STANDBY_TO_ACTIVE init tests for the SYNC outcome. In StoreAndForwardFailoverIT, gate transitionCluster2's peer-reaction settle-poll to the DEGRADED_STANDBY target (with named constants) and note that PHOENIX_REPLICATION_REPLAY_ENABLED also gates the compaction consistency-point guard. Document degradedListener's authoritative (non-CAS) set and the init-time compareAndSet(NOT_INITIALIZED, ...) yield-to-listener semantics. --- .../reader/ReplicationLogDiscoveryReplay.java | 57 +++++++++++--- .../ReplicationLogDiscoveryReplayTestIT.java | 77 +++++++++++++++++-- .../reader/StoreAndForwardFailoverIT.java | 34 +++++--- 3 files changed, 141 insertions(+), 27 deletions(-) diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java index 8cf469b7032..50f02bdd2cd 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java @@ -127,6 +127,13 @@ public void init() throws IOException { clusterType == ClusterType.LOCAL && HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(toState) ) { + // Unconditional set (not compareAndSet): degradation is an authoritative, fail-closed + // signal. Whatever the current state (SYNC, SYNCED_RECOVERY, or even NOT_INITIALIZED if a + // degrade lands mid-init), replay must stop advancing the sync point. There is no single + // valid "from" state to CAS against, so a CAS would silently drop legitimate degrade + // signals arriving from the other states. Contrast triggerFailoverListner below, which is + // deliberately conditional (compareAndSet from DEGRADED only) so it cannot clobber a + // healthy SYNC or an already-pending SYNCED_RECOVERY. replicationReplayState.set(ReplicationReplayState.DEGRADED); LOG.info("Cluster degraded detected for {}. replicationReplayState={}", haGroupName, ReplicationReplayState.DEGRADED); @@ -216,9 +223,11 @@ protected void processFile(Path path) throws IOException { * (lastRoundInSync from the minimum of lastSyncStateTimeInMs and the file frontier, so it * represents the last consistent point before degradation). *

  • STANDBY_TO_ACTIVE: a restart while already mid-failover (e.g. after a direct - * DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). Sets replicationReplayState to - * SYNCED_RECOVERY and initializes both rounds the same way, so the first replay() rewinds to - * lastRoundInSync before failover can promote.
    • + * DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). Initializes both rounds from the last + * known good sync point, then sets replicationReplayState to SYNCED_RECOVERY only when a rewind + * is actually pending (lastRoundInSync behind lastRoundProcessed) so the first replay() rewinds + * before failover can promote; when the rounds are already equal it initializes directly to SYNC + * so promotion is not delayed by a round window. *
  • Other states (e.g. STANDBY): sets replicationReplayState to SYNC, delegates to the parent * to initialize lastRoundProcessed, and sets lastRoundInSync equal to lastRoundProcessed.
    • * @@ -236,6 +245,11 @@ protected void initializeLastRoundProcessed() throws IOException { HAGroupStoreRecord.HAGroupState haGroupState = haGroupStoreRecord.getHAGroupState(); LOG.info("Found HA Group state during initialization as {} for haGroup: {}", haGroupState, haGroupName); + // Each branch below sets the initial state with compareAndSet(NOT_INITIALIZED, ...), not a + // plain set: the LOCAL state listeners are subscribed in init() before this runs, so a + // concurrent LOCAL transition may have already advanced the state off NOT_INITIALIZED. When + // that happens the CAS no-ops and we deliberately keep the listener's value -- a live + // transition is more recent (and thus more authoritative) than the record snapshot read above. if (HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupState)) { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.DEGRADED); @@ -245,12 +259,33 @@ protected void initializeLastRoundProcessed() throws IOException { // DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). No listener fires on a fresh process, so // initialize as if recovering: lastRoundInSync at the last good sync point, // lastRoundProcessed - // from the file frontier, and state SYNCED_RECOVERY so the first replay() rewinds before - // shouldTriggerFailover() can promote. Conservative by design: if we actually came from a - // healthy STANDBY, lastSyncStateTimeInMs ~= the file frontier and the rewind is a no-op. - replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, - ReplicationReplayState.SYNCED_RECOVERY); + // from the file frontier. initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime); + // Enter SYNCED_RECOVERY only when there is real rewind work (lastRoundInSync behind + // lastRoundProcessed) so the first replay() rewinds before shouldTriggerFailover() promotes. + // When the rounds are equal (the common "already fully replayed, RS bounced before + // setHAGroupStatusToSync() completed" case) there is nothing to rewind, so initialize + // directly to SYNC. Otherwise the SYNCED_RECOVERY -> SYNC CAS (which only fires inside the + // round-processing loop) would not run until a round becomes time-eligible, delaying + // promotion by ~one round window. + ReplicationReplayState initialState = + lastRoundInSync.getEndTime() < lastRoundProcessed.getEndTime() + ? ReplicationReplayState.SYNCED_RECOVERY + : ReplicationReplayState.SYNC; + // Distinct, greppable signal that this process restarted mid-failover, and which sub-case it + // hit, so a restart-driven promotion is diagnosable from logs (not inferred from the generic + // init summary below). + if (initialState == ReplicationReplayState.SYNCED_RECOVERY) { + LOG.info( + "Restart into STANDBY_TO_ACTIVE with rewind pending for haGroup {}: lastRoundInSync={} " + + "behind lastRoundProcessed={}; entering SYNCED_RECOVERY to re-sync before promotion.", + haGroupName, lastRoundInSync, lastRoundProcessed); + } else { + LOG.info("Restart into STANDBY_TO_ACTIVE with nothing to rewind for haGroup {}: " + + "lastRoundInSync == lastRoundProcessed ({}); entering SYNC, promotion eligible " + + "immediately.", haGroupName, lastRoundProcessed); + } + replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, initialState); } else { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.SYNC); @@ -263,9 +298,11 @@ protected void initializeLastRoundProcessed() throws IOException { + "replication replay state as {}", lastRoundProcessed, lastRoundInSync, replicationReplayState); - // Update the failoverPending variable during initialization + // Arm failoverPending during initialization when restarting mid-failover. Plain set (not + // compareAndSet): it starts false and this is the only initializer, so there is no prior value + // to guard; this also matches triggerFailoverListner, which arms the same flag with set(true). if (HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) { - failoverPending.compareAndSet(false, true); + failoverPending.set(true); } } 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 9eef0cd74cd..35f0036c653 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 @@ -466,8 +466,10 @@ public void testInitializeLastRoundProcessed_SyncStateWithNoFiles() throws IOExc } /** - * Tests initializeLastRoundProcessed in STANDBY_TO_ACTIVE state. Validates that failoverPending - * is set to true when HA group state is STANDBY_TO_ACTIVE. + * Tests initializeLastRoundProcessed in STANDBY_TO_ACTIVE state with no files. lastRoundInSync + * collapses onto lastRoundProcessed (nothing to rewind), so the replay initializes directly to + * SYNC (not SYNCED_RECOVERY) and promotion is not delayed by a round window. failoverPending is + * still armed because the HA group state is STANDBY_TO_ACTIVE. */ @Test public void testInitializeLastRoundProcessed_StandbyToActiveState() throws IOException { @@ -487,8 +489,7 @@ public void testInitializeLastRoundProcessed_StandbyToActiveState() throws IOExc testInitializeLastRoundProcessedHelper(currentTime, null, null, null, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, - expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, - true); + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, true); } @Test @@ -534,14 +535,74 @@ public void testInitializeLastRoundProcessed_StandbyToActiveStateWithZeroLastSyn new ReplicationRound(expectedEndTime - roundTimeMills, expectedEndTime); // With lastSyncStateTimeInMs == 0 (unset), lastRoundInSync must collapse onto the file frontier - // (== lastRoundProcessed) instead of ReplicationRound(0, 0), which would rewind SYNCED_RECOVERY - // all the way to the epoch and drive the consistency point to 0 (retain-everything). + // (== lastRoundProcessed) instead of ReplicationRound(0, 0), which would rewind all the way to + // the epoch and drive the consistency point to 0 (retain-everything). Because the two rounds + // are then equal (nothing to rewind), the replay initializes directly to SYNC rather than + // SYNCED_RECOVERY, so promotion is not delayed by a round window. ReplicationRound expectedLastRoundInSync = expectedLastRoundProcessed; testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, - expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY, - true); + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, true); + } + + /** + * PHOENIX-7920 restart into STANDBY_TO_ACTIVE with nothing left to replay: a fresh replay that + * reads a persisted STANDBY_TO_ACTIVE record (an RS bounced mid-failover after the data was + * already replayed) must promote on the FIRST replay() tick. With no files, lastRoundInSync + * collapses onto lastRoundProcessed, so init picks SYNC (not SYNCED_RECOVERY), failoverPending is + * armed, and shouldTriggerFailover() passes immediately. Regression guard for the init-branch + * latency fix: on the earlier unconditional-SYNCED_RECOVERY init, the SYNC state gate would block + * promotion until a round became time-eligible (~one round window later), so triggerFailover + * would not fire on this first tick. + */ + @Test + public void testReplay_RestartIntoStandbyToActive_NothingToReplay_PromotesOnFirstTick() + throws IOException { + long currentTime = 1704153600000L; + + TestableReplicationLogTracker fileTracker = + createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri); + fileTracker.init(); + try { + // Persisted STANDBY_TO_ACTIVE with no in / in-progress files: models an RS that restarted + // after a direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition once the data had already + // been replayed. + HAGroupStoreRecord record = + new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, haGroupName, + HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, currentTime, + HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl, CLUSTERS.getMasterAddress1(), + CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L); + + EnvironmentEdge edge = () -> currentTime; + EnvironmentEdgeManager.injectEdge(edge); + try { + TestableReplicationLogDiscoveryReplay discovery = + new TestableReplicationLogDiscoveryReplay(fileTracker, record); + // init() subscribes the LOCAL listeners (none fire here: this group's LOCAL state is + // ACTIVE, + // not one of the subscribed failover/degrade targets, and the record is never transitioned + // in this test), runs initializeLastRoundProcessed() over the injected STANDBY_TO_ACTIVE + // record, and wires up the metrics source that replay() needs. + discovery.init(); + + // Nothing to rewind -> init directly to SYNC with failoverPending armed. + assertEquals("no-rewind restart into STANDBY_TO_ACTIVE must init to SYNC", + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); + assertTrue("failoverPending must be armed on a STANDBY_TO_ACTIVE restart", + discovery.getFailoverPending()); + + // First replay() tick must promote (shouldTriggerFailover passes since state is SYNC). + discovery.replay(); + assertEquals("promotion must fire on the first replay tick", 1, + discovery.getTriggerFailoverCallCount()); + } finally { + EnvironmentEdgeManager.reset(); + } + } finally { + fileTracker.close(); + } } /** diff --git a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java index d3d5690cd8c..b03a018f6db 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java @@ -75,6 +75,12 @@ public class StoreAndForwardFailoverIT extends HABaseIT { private static final int B_START = 50; private static final int B_COUNT = 50; + // Cluster2's peer-aware reaction to Cluster1's ACTIVE_NOT_IN_SYNC drives its LOCAL record to + // DEGRADED_STANDBY asynchronously; bound how long transitionCluster2 waits for that reaction to + // land (and how often it polls) before attempting its own transition. + private static final long PEER_REACTION_SETTLE_MILLIS = 10000L; + private static final long PEER_REACTION_POLL_MILLIS = 500L; + @Rule public TestName name = new TestName(); @@ -98,6 +104,11 @@ public static synchronized void doSetup() throws Exception { // Disable the auto replay scheduler so our manually-driven instance is the only replayer. // (HABaseIT.doBaseSetup enables it; the writer/forwarder are gated on SYNCHRONOUS_REPLICATION, // not on this flag, so they are unaffected.) + // This flag has a second consumer: CompactionScanner also gates the replication + // consistency-point compaction guard on it, so setting it false disables that guard during + // major compactions too. Both effects share this one flag and cannot be separated. It is + // immaterial here: this test triggers no major compaction and asserts cross-cluster + // cell-equality directly, so the guard's absence does not affect what it verifies. conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); CLUSTERS.start(); @@ -282,17 +293,22 @@ private void stageEnterStoreAndForwardAndForwardBatchB() throws Exception { *

    * Cluster2 is peer-aware: when Cluster1 persists {@code ACTIVE_NOT_IN_SYNC} on entering * store-and-forward, Cluster2 auto-reacts and drives its own LOCAL record to - * {@code DEGRADED_STANDBY}. That reaction may still be in flight when this helper runs, so we - * first poll a bounded window for the record to settle at {@code target} and treat "already at - * target" as a no-op — otherwise a redundant transition (e.g. DEGRADED_STANDBY -> - * DEGRADED_STANDBY) would throw InvalidClusterRoleTransitionException on a slow box. + * {@code DEGRADED_STANDBY}. That reaction may still be in flight when this helper runs, so for + * the {@code DEGRADED_STANDBY} target only we first poll a bounded window for the record to + * settle and treat "already at target" as a no-op — otherwise a redundant transition (e.g. + * DEGRADED_STANDBY -> DEGRADED_STANDBY) would throw InvalidClusterRoleTransitionException on a + * slow box. Other targets (e.g. {@code STANDBY_TO_ACTIVE}) are never peer-driven, so the + * settle-poll is skipped for them to avoid dead time. */ private void transitionCluster2(HAGroupStoreRecord.HAGroupState target) throws Exception { - // Let any in-flight peer-aware reaction settle so the no-op check below fires - // deterministically. - long settleDeadline = System.currentTimeMillis() + 10000L; - while (System.currentTimeMillis() < settleDeadline && !cluster2StateIs(target)) { - Thread.sleep(500L); + // Only DEGRADED_STANDBY is reached by the peer-aware auto-reaction; let any in-flight reaction + // settle so the no-op check below fires deterministically. Skipping this for non-peer-driven + // targets avoids waiting on a condition that can never become true. + if (target == HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY) { + long settleDeadline = System.currentTimeMillis() + PEER_REACTION_SETTLE_MILLIS; + while (System.currentTimeMillis() < settleDeadline && !cluster2StateIs(target)) { + Thread.sleep(PEER_REACTION_POLL_MILLIS); + } } // Check if already at target - no-op if so (avoid redundant transition exception). if (cluster2StateIs(target)) {