Phoenix 7920 Add replay-side handling for direct DEGRADED_STANDBY → STANDBY_TO_ACTIVE transition (ANISTS → AISTS)

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<Path> processOneRandomFile(final List<Path> 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<Long> 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);
       }
     }
   }

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.
+   * <ul>
+   * <li>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).</li>
+   * <li>STANDBY_TO_ACTIVE: a restart while already mid-failover (e.g. after a direct
+   * DEGRADED_STANDBY -&gt; 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.</li>
+   * <li>Other states (e.g. STANDBY): sets replicationReplayState to SYNC, delegates to the parent
+   * to initialize lastRoundProcessed, and sets lastRoundInSync equal to lastRoundProcessed.</li>
+   * </ul>
+   * 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<Long> minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles();
-      Optional<Long> 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<Long> minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles();
+    Optional<Long> 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

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<Path> findLogFiles(Path dir, FileSystem fs) throws IOException {
+    List<Path> files = new ArrayList<>();
+    findLogFilesRecursive(dir, fs, files);
+    return files;
+  }
+
+  private static void findLogFilesRecursive(Path dir, FileSystem fs, List<Path> 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);
+      }
+    }
+  }
+}