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..638d95280f5 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java @@ -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,30 @@ 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 +426,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..50f02bdd2cd 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java @@ -127,6 +127,13 @@ public void init() throws IOException { clusterType == ClusterType.LOCAL && HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(toState) ) { + // Unconditional set (not compareAndSet): degradation is an authoritative, fail-closed + // signal. Whatever the current state (SYNC, SYNCED_RECOVERY, or even NOT_INITIALIZED if a + // degrade lands mid-init), replay must stop advancing the sync point. There is no single + // valid "from" state to CAS against, so a CAS would silently drop legitimate degrade + // signals arriving from the other states. Contrast triggerFailoverListner below, which is + // deliberately conditional (compareAndSet from DEGRADED only) so it cannot clobber a + // healthy SYNC or an already-pending SYNCED_RECOVERY. replicationReplayState.set(ReplicationReplayState.DEGRADED); LOG.info("Cluster degraded detected for {}. replicationReplayState={}", haGroupName, ReplicationReplayState.DEGRADED); @@ -151,6 +158,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 +216,80 @@ 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). Initializes both rounds from the last + * known good sync point, then sets replicationReplayState to SYNCED_RECOVERY only when a rewind + * is actually pending (lastRoundInSync behind lastRoundProcessed) so the first replay() rewinds + * before failover can promote; when the rounds are already equal it initializes directly to SYNC + * so promotion is not delayed by a round window.
    • + *
  • Other states (e.g. STANDBY): sets replicationReplayState to SYNC, delegates to the parent + * to initialize lastRoundProcessed, and sets lastRoundInSync equal to lastRoundProcessed.
    • + *
+ * 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); + // Each branch below sets the initial state with compareAndSet(NOT_INITIALIZED, ...), not a + // plain set: the LOCAL state listeners are subscribed in init() before this runs, so a + // concurrent LOCAL transition may have already advanced the state off NOT_INITIALIZED. When + // that happens the CAS no-ops and we deliberately keep the listener's value -- a live + // transition is more recent (and thus more authoritative) than the record snapshot read above. + if (HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupState)) { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.DEGRADED); - 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()); + 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. + initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime); + // Enter SYNCED_RECOVERY only when there is real rewind work (lastRoundInSync behind + // lastRoundProcessed) so the first replay() rewinds before shouldTriggerFailover() promotes. + // When the rounds are equal (the common "already fully replayed, RS bounced before + // setHAGroupStatusToSync() completed" case) there is nothing to rewind, so initialize + // directly to SYNC. Otherwise the SYNCED_RECOVERY -> SYNC CAS (which only fires inside the + // round-processing loop) would not run until a round becomes time-eligible, delaying + // promotion by ~one round window. + ReplicationReplayState initialState = + lastRoundInSync.getEndTime() < lastRoundProcessed.getEndTime() + ? ReplicationReplayState.SYNCED_RECOVERY + : ReplicationReplayState.SYNC; + // Distinct, greppable signal that this process restarted mid-failover, and which sub-case it + // hit, so a restart-driven promotion is diagnosable from logs (not inferred from the generic + // init summary below). + if (initialState == ReplicationReplayState.SYNCED_RECOVERY) { + LOG.info( + "Restart into STANDBY_TO_ACTIVE with rewind pending for haGroup {}: lastRoundInSync={} " + + "behind lastRoundProcessed={}; entering SYNCED_RECOVERY to re-sync before promotion.", + haGroupName, lastRoundInSync, lastRoundProcessed); + } else { + LOG.info("Restart into STANDBY_TO_ACTIVE with nothing to rewind for haGroup {}: " + + "lastRoundInSync == lastRoundProcessed ({}); entering SYNC, promotion eligible " + + "immediately.", haGroupName, lastRoundProcessed); } - 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)); + replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, initialState); } else { replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, ReplicationReplayState.SYNC); - super.initializeLastRoundProcessed(); + super.initializeLastRoundProcessed(frontierStartTime); this.lastRoundInSync = new ReplicationRound(lastRoundProcessed.getStartTime(), lastRoundProcessed.getEndTime()); } @@ -254,12 +298,56 @@ protected void initializeLastRoundProcessed() throws IOException { + "replication replay state as {}", lastRoundProcessed, lastRoundInSync, replicationReplayState); - // Update the failoverPending variable during initialization - if ( - HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupStoreRecord.getHAGroupState()) - ) { - failoverPending.compareAndSet(false, true); + // Arm failoverPending during initialization when restarting mid-failover. Plain set (not + // compareAndSet): it starts false and this is the only initializer, so there is no prior value + // to guard; this also matches triggerFailoverListner, which arms the same flag with set(true). + if (HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) { + failoverPending.set(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); } /** 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..127f793415f --- /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..335cdee1adc 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/ReplicationLogGroupIT.java @@ -24,35 +24,26 @@ import static org.apache.phoenix.query.BaseTest.generateUniqueName; import static org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; import java.io.IOException; import java.sql.Connection; 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; @@ -244,7 +235,7 @@ private void replayAndVerifyAcrossClusters(List ddlStatements, String... // Replay replication log on cluster 2 FileSystem fs = standByLogDir.getFileSystem(conf2); - List logFiles = findLogFiles(standByLogDir, fs); + List logFiles = CrossClusterReplicationTestUtil.findLogFiles(standByLogDir, fs); assertTrue("Should have at least one log file", !logFiles.isEmpty()); ReplicationLogProcessor processor = ReplicationLogProcessor.get(conf2, haGroupName); try { @@ -721,65 +712,7 @@ public void testSystemTables() throws Exception { assertTrue(getCountForTable(systemTables, SYSTEM_CATALOG_NAME) > 0); } - 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()); - } - } - } - 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..1b363c8b0ee --- /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..35f0036c653 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java @@ -350,6 +350,32 @@ 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; @@ -440,8 +466,10 @@ public void testInitializeLastRoundProcessed_SyncStateWithNoFiles() throws IOExc } /** - * Tests initializeLastRoundProcessed in STANDBY_TO_ACTIVE state. Validates that failoverPending - * is set to true when HA group state is STANDBY_TO_ACTIVE. + * Tests initializeLastRoundProcessed in STANDBY_TO_ACTIVE state with no files. lastRoundInSync + * collapses onto lastRoundProcessed (nothing to rewind), so the replay initializes directly to + * SYNC (not SYNCED_RECOVERY) and promotion is not delayed by a round window. failoverPending is + * still armed because the HA group state is STANDBY_TO_ACTIVE. */ @Test public void testInitializeLastRoundProcessed_StandbyToActiveState() throws IOException { @@ -464,6 +492,119 @@ public void testInitializeLastRoundProcessed_StandbyToActiveState() throws IOExc expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, 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 all the way to + // the epoch and drive the consistency point to 0 (retain-everything). Because the two rounds + // are then equal (nothing to rewind), the replay initializes directly to SYNC rather than + // SYNCED_RECOVERY, so promotion is not delayed by a round window. + ReplicationRound expectedLastRoundInSync = expectedLastRoundProcessed; + + testInitializeLastRoundProcessedHelper(currentTime, lastSyncStateTime, newFileTimestamp, null, + HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, expectedLastRoundProcessed, + expectedLastRoundInSync, ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, true); + } + + /** + * PHOENIX-7920 restart into STANDBY_TO_ACTIVE with nothing left to replay: a fresh replay that + * reads a persisted STANDBY_TO_ACTIVE record (an RS bounced mid-failover after the data was + * already replayed) must promote on the FIRST replay() tick. With no files, lastRoundInSync + * collapses onto lastRoundProcessed, so init picks SYNC (not SYNCED_RECOVERY), failoverPending is + * armed, and shouldTriggerFailover() passes immediately. Regression guard for the init-branch + * latency fix: on the earlier unconditional-SYNCED_RECOVERY init, the SYNC state gate would block + * promotion until a round became time-eligible (~one round window later), so triggerFailover + * would not fire on this first tick. + */ + @Test + public void testReplay_RestartIntoStandbyToActive_NothingToReplay_PromotesOnFirstTick() + throws IOException { + long currentTime = 1704153600000L; + + TestableReplicationLogTracker fileTracker = + createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri); + fileTracker.init(); + try { + // Persisted STANDBY_TO_ACTIVE with no in / in-progress files: models an RS that restarted + // after a direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition once the data had already + // been replayed. + HAGroupStoreRecord record = + new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, haGroupName, + HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE, currentTime, + HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl, CLUSTERS.getMasterAddress1(), + CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L); + + EnvironmentEdge edge = () -> currentTime; + EnvironmentEdgeManager.injectEdge(edge); + try { + TestableReplicationLogDiscoveryReplay discovery = + new TestableReplicationLogDiscoveryReplay(fileTracker, record); + // init() subscribes the LOCAL listeners (none fire here: this group's LOCAL state is + // ACTIVE, + // not one of the subscribed failover/degrade targets, and the record is never transitioned + // in this test), runs initializeLastRoundProcessed() over the injected STANDBY_TO_ACTIVE + // record, and wires up the metrics source that replay() needs. + discovery.init(); + + // Nothing to rewind -> init directly to SYNC with failoverPending armed. + assertEquals("no-rewind restart into STANDBY_TO_ACTIVE must init to SYNC", + ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC, + discovery.getReplicationReplayState()); + assertTrue("failoverPending must be armed on a STANDBY_TO_ACTIVE restart", + discovery.getFailoverPending()); + + // First replay() tick must promote (shouldTriggerFailover passes since state is SYNC). + discovery.replay(); + assertEquals("promotion must fire on the first replay tick", 1, + discovery.getTriggerFailoverCallCount()); + } finally { + EnvironmentEdgeManager.reset(); + } + } finally { + fileTracker.close(); + } + } + /** * Helper method to test initializeLastRoundProcessed with various file and state configurations. * Handles file creation, state setup, and validation of lastRoundProcessed and lastRoundInSync. @@ -1260,6 +1401,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..b03a018f6db --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/StoreAndForwardFailoverIT.java @@ -0,0 +1,395 @@ +/* + * 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; + + // Cluster2's peer-aware reaction to Cluster1's ACTIVE_NOT_IN_SYNC drives its LOCAL record to + // DEGRADED_STANDBY asynchronously; bound how long transitionCluster2 waits for that reaction to + // land (and how often it polls) before attempting its own transition. + private static final long PEER_REACTION_SETTLE_MILLIS = 10000L; + private static final long PEER_REACTION_POLL_MILLIS = 500L; + + @Rule + public TestName name = new TestName(); + + 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.) + // This flag has a second consumer: CompactionScanner also gates the replication + // consistency-point compaction guard on it, so setting it false disables that guard during + // major compactions too. Both effects share this one flag and cannot be separated. It is + // immaterial here: this test triggers no major compaction and asserts cross-cluster + // cell-equality directly, so the guard's absence does not affect what it verifies. + conf1.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); + conf2.setBoolean(PHOENIX_REPLICATION_REPLAY_ENABLED, false); + CLUSTERS.start(); + } + + @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 for + * the {@code DEGRADED_STANDBY} target only we first poll a bounded window for the record to + * settle and treat "already at target" as a no-op — otherwise a redundant transition (e.g. + * DEGRADED_STANDBY -> DEGRADED_STANDBY) would throw InvalidClusterRoleTransitionException on a + * slow box. Other targets (e.g. {@code STANDBY_TO_ACTIVE}) are never peer-driven, so the + * settle-poll is skipped for them to avoid dead time. + */ + private void transitionCluster2(HAGroupStoreRecord.HAGroupState target) throws Exception { + // Only DEGRADED_STANDBY is reached by the peer-aware auto-reaction; let any in-flight reaction + // settle so the no-op check below fires deterministically. Skipping this for non-peer-driven + // targets avoids waiting on a condition that can never become true. + if (target == HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY) { + long settleDeadline = System.currentTimeMillis() + PEER_REACTION_SETTLE_MILLIS; + while (System.currentTimeMillis() < settleDeadline && !cluster2StateIs(target)) { + Thread.sleep(PEER_REACTION_POLL_MILLIS); + } + } + // Check if already at target - no-op if so (avoid redundant transition exception). + if (cluster2StateIs(target)) { + 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"); + } +}