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/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/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 { 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"); + } +}