From 22eeb7e82cce74ca62816ed2e132340c10df39bb Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 9 Jul 2026 16:41:18 +0530 Subject: [PATCH 01/13] PHOENIX-7920 triggerFailoverListner rewinds DEGRADED->SYNCED_RECOVERY on direct failover --- .../reader/ReplicationLogDiscoveryReplay.java | 7 + .../ReplicationLogDiscoveryReplayTestIT.java | 168 ++++++++++++++++++ 2 files changed, 175 insertions(+) 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..0f36295cfd8 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={}. " 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..1e2b981ef7f 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 @@ -1260,6 +1260,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 da4c6818e366a7dfd014eeefdba69d95cd962601 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 9 Jul 2026 17:14:03 +0530 Subject: [PATCH 02/13] PHOENIX-7920 init STANDBY_TO_ACTIVE as SYNCED_RECOVERY from last sync point --- .../reader/ReplicationLogDiscoveryReplay.java | 96 ++++++++++++------- .../ReplicationLogDiscoveryReplayTestIT.java | 31 +++++- 2 files changed, 91 insertions(+), 36 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 0f36295cfd8..583b63c5275 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 @@ -209,46 +209,43 @@ 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. + * + * 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); 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); + } 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); } else { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.SYNC); @@ -262,13 +259,42 @@ 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 current time 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. + */ + private void initLastRoundsFromLastSyncPoint(HAGroupStoreRecord haGroupStoreRecord) + throws IOException { + 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)); + } + /** * 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 1e2b981ef7f..b37cbf1a7e0 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 @@ -461,7 +461,36 @@ 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); } /** From a09d15924a09b1d762da4c558ed705af20c540bc Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 9 Jul 2026 17:29:12 +0530 Subject: [PATCH 03/13] PHOENIX-7920 make lastRoundProcessed volatile for visibility parity --- .../org/apache/phoenix/replication/ReplicationLogDiscovery.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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..d2bed44b92c 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; From 19686bbf5ca6060065a802490628cd4a3b8c8e82 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Sat, 11 Jul 2026 14:55:55 +0530 Subject: [PATCH 04/13] PHOENIX-7920 guard lastRoundInSync against zero lastSyncStateTimeInMs (avoid epoch rewind) --- .../reader/ReplicationLogDiscoveryReplay.java | 14 ++++++++--- .../ReplicationLogDiscoveryReplayTestIT.java | 25 +++++++++++++++++++ 2 files changed, 36 insertions(+), 3 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 583b63c5275..83cbf36aa7e 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 @@ -290,9 +290,17 @@ private void initLastRoundsFromLastSyncPoint(HAGroupStoreRecord haGroupStoreReco } this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager() .getReplicationRoundFromEndTime(minimumTimestampFromFiles); - this.lastRoundInSync = - replicationLogTracker.getReplicationShardDirectoryManager().getReplicationRoundFromEndTime( - Math.min(haGroupStoreRecord.getLastSyncStateTimeInMs(), 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); } /** 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 b37cbf1a7e0..59a72152ee7 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 @@ -493,6 +493,31 @@ public void testInitializeLastRoundProcessed_StandbyToActiveStateWithLastSyncSta 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); + } + /** * Helper method to test initializeLastRoundProcessed with various file and state configurations. * Handles file creation, state setup, and validation of lastRoundProcessed and lastRoundInSync. From 9a3d90c8e6f4638f70bda39160c5dfc6de85c726 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Sat, 11 Jul 2026 16:10:35 +0530 Subject: [PATCH 05/13] PHOENIX-7920 Sample init frontier time before reading HA group state Capture current time at the start of initializeLastRoundProcessed() (before the HA group state read and file scans) and thread it through all three init branches - DEGRADED_STANDBY, STANDBY_TO_ACTIVE, and the SYNC path - via a new fallback-time overload on the parent. This anchors the starting round to when init began, so a SYNC -> DEGRADED_STANDBY transition during init cannot push the round (and the SYNC-path recovery floor) past the degradation point. Also documents the zero lastSyncStateTimeInMs fallback on initLastRoundsFromLastSyncPoint. --- .../replication/ReplicationLogDiscovery.java | 24 ++++++++++++----- .../reader/ReplicationLogDiscoveryReplay.java | 27 ++++++++++++------- 2 files changed, 36 insertions(+), 15 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 d2bed44b92c..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 @@ -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 83cbf36aa7e..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 @@ -228,6 +228,10 @@ protected void processFile(Path path) throws IOException { @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(); HAGroupStoreRecord.HAGroupState haGroupState = haGroupStoreRecord.getHAGroupState(); LOG.info("Found HA Group state during initialization as {} for haGroup: {}", haGroupState, @@ -235,7 +239,7 @@ protected void initializeLastRoundProcessed() throws IOException { if (HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupState)) { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.DEGRADED); - initLastRoundsFromLastSyncPoint(haGroupStoreRecord); + 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 @@ -245,11 +249,11 @@ protected void initializeLastRoundProcessed() throws IOException { // healthy STANDBY, lastSyncStateTimeInMs ~= the file frontier and the rewind is a no-op. replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.SYNCED_RECOVERY); - initLastRoundsFromLastSyncPoint(haGroupStoreRecord); + initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime); } else { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.SYNC); - super.initializeLastRoundProcessed(); + super.initializeLastRoundProcessed(frontierStartTime); this.lastRoundInSync = new ReplicationRound(lastRoundProcessed.getStartTime(), lastRoundProcessed.getEndTime()); } @@ -268,13 +272,18 @@ protected void initializeLastRoundProcessed() throws IOException { * 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 current time 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. + * 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) - throws IOException { - long minimumTimestampFromFiles = EnvironmentEdgeManager.currentTime(); + private void initLastRoundsFromLastSyncPoint(HAGroupStoreRecord haGroupStoreRecord, + long frontierStartTime) throws IOException { + long minimumTimestampFromFiles = frontierStartTime; Optional minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles(); Optional minTimestampFromNewFiles = getMinTimestampFromNewFiles(); if (minTimestampFromInProgressFiles.isPresent()) { From e24ab1bec198c60f7bf5c0a1621ea6cef0a52f28 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Mon, 13 Jul 2026 12:29:01 +0530 Subject: [PATCH 06/13] PHOENIX-7920 Add DEGRADED_STANDBY zero-lastSyncStateTimeInMs regression test The zero-lastSyncStateTimeInMs guard in initLastRoundsFromLastSyncPoint is shared by the DEGRADED_STANDBY and STANDBY_TO_ACTIVE init paths, but only the STANDBY_TO_ACTIVE path had a zero-sync regression test. The guard also corrected the DEGRADED path (which previously computed round(min(0, frontier)) = round(0) = ReplicationRound(0,0), the epoch-rewind bug). This adds the symmetric DEGRADED_STANDBY test so a future refactor that diverges the two paths cannot silently reintroduce the epoch rewind. Test count 44 -> 45 (failsafe XML verified GREEN). --- .../ReplicationLogDiscoveryReplayTestIT.java | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) 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 59a72152ee7..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; From 5d8834a82b975d9002b7a751fb05d386cbb224d7 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Wed, 15 Jul 2026 21:45:23 +0530 Subject: [PATCH 07/13] PHOENIX-7920 Add SAF-failover test scaffolding: mode-flip shim + shared cross-cluster helpers --- .../CrossClusterReplicationTestUtil.java | 117 ++++++++++++++++++ .../replication/ReplicationLogGroupIT.java | 65 +--------- .../ReplicationLogGroupTestAccess.java | 67 ++++++++++ 3 files changed, 186 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 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(); + } +} From 3a075e3cd2beaf93df3d17564f23488c40e2819b Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Wed, 15 Jul 2026 21:56:10 +0530 Subject: [PATCH 08/13] PHOENIX-7920 SAF failover E2E: harness skeleton, real writer + manually-driven replay, SYNC init --- .../reader/StoreAndForwardFailoverIT.java | 212 ++++++++++++++++++ 1 file changed, 212 insertions(+) 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/reader/StoreAndForwardFailoverIT.java b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java new file mode 100644 index 00000000000..08fa1d882fb --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java @@ -0,0 +1,212 @@ +/* + * 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.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.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). See the + * design spec: docs/superpowers/specs/2026-07-15-phoenix-7920-store-and-forward-failover-e2e-design.md + */ +@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; + + @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()); + // Stages 1-3 are added in Tasks 3-5. + } + + /** 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(); + } + } +} From 7dc88408fa1bf57fe35af2cdf6b713c92e68e002 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Wed, 15 Jul 2026 22:04:31 +0530 Subject: [PATCH 09/13] PHOENIX-7920 SAF failover E2E: stage 1 - SYNC batch A replicates and advances sync point --- .../reader/StoreAndForwardFailoverIT.java | 32 ++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) 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 08fa1d882fb..fb3ea23aaf1 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 @@ -43,6 +43,7 @@ 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; @@ -86,6 +87,7 @@ public class StoreAndForwardFailoverIT extends HABaseIT { private FileSystem standbyFs; private Path standbyInDir; private String tableName; + private long syncPointAfterA; @BeforeClass public static synchronized void doSetup() throws Exception { @@ -163,7 +165,9 @@ public void testDirectFailoverInStoreAndForwardModeIsZeroRPO() throws Exception assertEquals("replay must initialize in SYNC while cluster2 is STANDBY", ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, discovery.getReplicationReplayState()); - // Stages 1-3 are added in Tasks 3-5. + // Stage 1: SYNC batch A replicates to cluster2 and advances the sync point. + stageWriteAndReplayBatchA(); + // Stages 2-3 are added in Tasks 4-5. } /** Poll until {@code condition} holds or {@code timeoutMs} elapses, then assert it. */ @@ -209,4 +213,30 @@ private void upsertRows(int startId, int count) throws SQLException { 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); + } } From 1bf9d29299a0b0e620aa9a32e0ad35ac05179cb9 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Wed, 15 Jul 2026 22:33:01 +0530 Subject: [PATCH 10/13] PHOENIX-7920 SAF failover E2E: stage 2 - enter store-and-forward and forward batch B --- .../reader/StoreAndForwardFailoverIT.java | 33 ++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) 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 fb3ea23aaf1..37976dedfe2 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 @@ -88,6 +88,7 @@ public class StoreAndForwardFailoverIT extends HABaseIT { private Path standbyInDir; private String tableName; private long syncPointAfterA; + private int logFileCountAfterA; @BeforeClass public static synchronized void doSetup() throws Exception { @@ -167,7 +168,9 @@ public void testDirectFailoverInStoreAndForwardModeIsZeroRPO() throws Exception discovery.getReplicationReplayState()); // Stage 1: SYNC batch A replicates to cluster2 and advances the sync point. stageWriteAndReplayBatchA(); - // Stages 2-3 are added in Tasks 4-5. + // Stage 2: enter store-and-forward, write batch B, await it forwarded to cluster2 'in'. + stageEnterStoreAndForwardAndForwardBatchB(); + // Stage 3 is added in Task 5. } /** Poll until {@code condition} holds or {@code timeoutMs} elapses, then assert it. */ @@ -239,4 +242,32 @@ private void stageWriteAndReplayBatchA() throws Exception { 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'"); + } } From 16ec5b51ba19ae527e0d071ae4c91faaf6ead772 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Wed, 15 Jul 2026 22:42:51 +0530 Subject: [PATCH 11/13] PHOENIX-7920 SAF failover E2E: stage 3 - direct failover promotes with zero RPO --- .../reader/StoreAndForwardFailoverIT.java | 89 ++++++++++++++++++- 1 file changed, 88 insertions(+), 1 deletion(-) 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 37976dedfe2..2352ea1a50f 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 @@ -38,6 +38,7 @@ 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; @@ -170,7 +171,8 @@ public void testDirectFailoverInStoreAndForwardModeIsZeroRPO() throws Exception stageWriteAndReplayBatchA(); // Stage 2: enter store-and-forward, write batch B, await it forwarded to cluster2 'in'. stageEnterStoreAndForwardAndForwardBatchB(); - // Stage 3 is added in Task 5. + // 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. */ @@ -270,4 +272,89 @@ private void stageEnterStoreAndForwardAndForwardBatchB() throws Exception { }, 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. + * If already at the target state, this is a no-op. + */ + private void transitionCluster2(HAGroupStoreRecord.HAGroupState target) throws Exception { + // 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"); + + // Drive replay: SYNCED_RECOVERY rewinds to lastRoundInSync (A), re-replays B in SYNC, then + // 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 c2f06e531d5261ba78da94882bcad3fb989b9d86 Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 16 Jul 2026 00:14:00 +0530 Subject: [PATCH 12/13] PHOENIX-7920 SAF failover E2E: harden transitionCluster2 against peer-aware race Cluster2 is peer-aware and auto-reacts to Cluster1's ACTIVE_NOT_IN_SYNC by driving its own LOCAL record to DEGRADED_STANDBY during store-and-forward entry. That reaction can still be in flight when transitionCluster2(DEGRADED_STANDBY) runs on a slow box, so setHAGroupStatusIfNeeded could throw InvalidClusterRoleTransitionException on a redundant same-state transition. Poll a bounded 10s window for the record to settle at the target before the no-op check so the transition is robust whether or not the reaction has landed. --- .../reader/StoreAndForwardFailoverIT.java | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) 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 2352ea1a50f..1c1724e26c3 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 @@ -277,9 +277,20 @@ 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. - * If already at the target state, this is a no-op. + * + *

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; From 77e02312e6089f1a0e1139c0f178fe0b32a5bd9e Mon Sep 17 00:00:00 2001 From: Himanshu Gwalani Date: Thu, 16 Jul 2026 12:42:17 +0530 Subject: [PATCH 13/13] PHOENIX-7920 SAF failover E2E: quiesce writer before promotion drive to fix flaky failover The promotion poll (driveReplayUntil ACTIVE_IN_SYNC) was flaky on slow/loaded boxes: it timed out at 120s on a 176.8s run but passed at ~57s runs. Root cause is not the guarded Fix #1 code but the test harness. Cluster1's writer was left running through Stage 3; its rotation thread emits an empty round .plog every round-duration into Cluster2's 'in' dir. That file perpetually occupies the [nextRoundToProcess, currentTimestampRound] window that shouldTriggerFailover() condition 4 (getNewFiles(...).isEmpty()) scans, so promotion could only land in a race window between the replay catching up and the next file arriving. The failed run shows 239/239 checks blocked with 'New files exist...'. Fix: close logGroup before the promotion drive, modeling the real failover precondition (the old ACTIVE is gone, producing no more logs). File production stops, the replay drains, condition 4 clears, and promotion is deterministic at any round duration. Batch B is already forwarded (Stage 2), so the SYNCED_RECOVERY rewind still re-replays it and zero RPO holds; close() is idempotent so afterTest's close is a no-op. Also drop the local design-spec path from the class javadoc (spec is not being pushed). Verified: Tests run: 1, Failures: 0, Time elapsed 57.61s; log shows promotion firing ~2s after writer close via 'Failover can be triggered'. --- .../replication/reader/StoreAndForwardFailoverIT.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) 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 1c1724e26c3..9d487896d8d 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 @@ -63,8 +63,7 @@ /** * 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). See the - * design spec: docs/superpowers/specs/2026-07-15-phoenix-7920-store-and-forward-failover-e2e-design.md + * manually-driven real ReplicationLogDiscoveryReplay (the auto-scheduler is disabled). */ @Category(NeedsOwnMiniClusterTest.class) public class StoreAndForwardFailoverIT extends HABaseIT { @@ -354,8 +353,11 @@ private void stageDegradeDirectFailoverAndAssertZeroRPO() throws Exception { awaitCondition(() -> cluster2StateIs(HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE), 30000L, "cluster2 record should reflect STANDBY_TO_ACTIVE before driving failover"); - // Drive replay: SYNCED_RECOVERY rewinds to lastRoundInSync (A), re-replays B in SYNC, then - // shouldTriggerFailover() passes and triggerFailover() sets ACTIVE_IN_SYNC on Cluster2. + 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");