PHOENIX-7938 Round-align the sync-replication replay consistency point

phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java

@@ -555,7 +555,15 @@ public enum ReplicationReplayState {
   /**
    * Returns the consistency point timestamp based on the current replication replay state. The
    * consistency point in a standby cluster is defined as the timestamp such that all mutations
-   * whose timestamp less than this consistency point timestamp have been replayed
+   * whose timestamp is less than this consistency point timestamp have been replayed.
+   * <p>
+   * In SYNC state with files in progress, the minimum IN-PROGRESS timestamp is aligned down to
+   * the start time of the round it belongs to. Files within a round are moved to IN-PROGRESS in
+   * random order, so the minimum IN-PROGRESS timestamp may not be the oldest file of its round
+   * (an older sibling can still be waiting in the IN directory). Every file of a round has a
+   * timestamp greater than or equal to the round start, and earlier rounds are fully replayed
+   * first, so the round start is a safe exclusive upper bound that never advances past unreplayed
+   * files - without listing the IN directories.
    * @return The consistency point timestamp in milliseconds
    * @throws IOException if the consistency point cannot be determined based on current state
    */
@@ -571,8 +579,12 @@ public long getConsistencyPoint() throws IOException {
         Optional<Long> optionalMinTimestampInProgressTimestamp =
           getMinTimestampFromInProgressFiles();
         if (optionalMinTimestampInProgressTimestamp.isPresent()) {
-          // Use minimum timestamp from in-progress files as consistency point
-          consistencyPoint = optionalMinTimestampInProgressTimestamp.get();
+          // Align the minimum in-progress timestamp down to the start of the round it belongs to,
+          // so the consistency point never advances past older, still-unreplayed files of the same
+          // round that are waiting in the IN directory (files are picked in random order).
+          long minTimestampInProgress = optionalMinTimestampInProgressTimestamp.get();
+          consistencyPoint = replicationLogTracker.getReplicationShardDirectoryManager()
+            .getReplicationRoundFromStartTime(minTimestampInProgress).getStartTime();
         } else if (lastRoundInSync != null) {
           // Use lastRoundInSync end time if no in-progress files
           // Since we are in sync mode, both lastRoundProcessed and lastRoundInSync would be same.

phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java

@@ -653,6 +653,65 @@ public void testReplay_SyncState_ProcessMultipleRounds() throws IOException {
     }
   }
 
+  /**
+   * PHOENIX-7938: files within a round are moved to IN-PROGRESS in random order, so the minimum
+   * IN-PROGRESS timestamp may not be the oldest file of its round (an older sibling can still be
+   * waiting in the IN directory). The SYNC-state consistency point must therefore align the minimum
+   * IN-PROGRESS timestamp down to its round start rather than use it raw. Verified end-to-end
+   * against the real tracker and filesystem.
+   */
+  @Test
+  public void testGetConsistencyPoint_SyncState_AlignsInProgressMinToRoundStart()
+    throws IOException {
+    TestableReplicationLogTracker fileTracker =
+      createReplicationLogTracker(conf1, haGroupName, rootFs, rootUri);
+
+    try {
+      long roundTimeMills =
+        fileTracker.getReplicationShardDirectoryManager().getReplicationRoundDurationSeconds()
+          * 1000L;
+      long roundStart = 1704153600000L; // aligned to a round boundary (2024-01-02 00:00:00)
+      long inProgressFileTs = roundStart + 30000L; // 30s into the round, off the round boundary
+      ReplicationRound roundN = new ReplicationRound(roundStart, roundStart + roundTimeMills);
+
+      // Seed a single IN-PROGRESS file 30s into the round. getConsistencyPoint() reads only the
+      // IN-PROGRESS directory, so this is the minimum timestamp; the fix must align it down to the
+      // round start rather than expose it raw, because an older sibling from the same round could
+      // still be waiting in the IN directory (files are moved to IN-PROGRESS in random order).
+      Path inProgressDir = fileTracker.getInProgressDirPath();
+      rootFs.mkdirs(inProgressDir);
+      rootFs.create(new Path(inProgressDir, inProgressFileTs + "_rs-1_uuid.plog"), true).close();
+
+      long currentTime = roundStart + (2 * roundTimeMills);
+      EnvironmentEdge edge = () -> currentTime;
+      EnvironmentEdgeManager.injectEdge(edge);
+
+      try {
+        HAGroupStoreRecord mockRecord =
+          new HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, haGroupName,
+            HAGroupStoreRecord.HAGroupState.STANDBY, roundStart,
+            HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl, CLUSTERS.getMasterAddress1(),
+            CLUSTERS.getMasterAddress2(), CLUSTERS.getHdfsUrl1(), CLUSTERS.getHdfsUrl2(), 0L);
+
+        TestableReplicationLogDiscoveryReplay discovery =
+          new TestableReplicationLogDiscoveryReplay(fileTracker, mockRecord);
+        discovery.init();
+        discovery.setLastRoundProcessed(roundN);
+        discovery.setLastRoundInSync(roundN);
+        discovery
+          .setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC);
+
+        assertEquals(
+          "Consistency point must align to round start, not the raw min IN-PROGRESS timestamp",
+          roundStart, discovery.getConsistencyPoint());
+      } finally {
+        EnvironmentEdgeManager.reset();
+      }
+    } finally {
+      fileTracker.close();
+    }
+  }
+
   /**
    * Tests replay in DEGRADED state processing multiple rounds. Validates that lastRoundProcessed
    * advances but lastRoundInSync is preserved.

phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayConsistencyPointTest.java

@@ -0,0 +1,150 @@
+/*
+ * 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.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.net.URI;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.phoenix.replication.ReplicationLogTracker;
+import org.apache.phoenix.replication.ReplicationRound;
+import org.apache.phoenix.replication.ReplicationShardDirectoryManager;
+import org.apache.phoenix.replication.metrics.MetricsReplicationLogTracker;
+import org.apache.phoenix.replication.metrics.MetricsReplicationLogTrackerReplayImpl;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Unit tests for {@link ReplicationLogDiscoveryReplay#getConsistencyPoint()} focused on
+ * round alignment of the minimum IN-PROGRESS timestamp (PHOENIX-7938). Runs entirely on the
+ * local filesystem; no mini cluster required.
+ */
+public class ReplicationLogDiscoveryReplayConsistencyPointTest {
+
+  private static final String HA_GROUP_NAME = "testGroup";
+  // Derive from the production default instead of hardcoding, so this tracks the real round
+  // duration if the default changes. ROUND_START is aligned to a round boundary.
+  private static final long ROUND_MILLIS =
+    ReplicationShardDirectoryManager.DEFAULT_REPLICATION_ROUND_DURATION_SECONDS * 1000L;
+  private static final long ROUND_START = 1704153600000L; // divisible by 60000
+
+  @Rule
+  public TemporaryFolder testFolder = new TemporaryFolder();
+
+  private Configuration conf;
+  private FileSystem localFs;
+  private TestableTracker tracker;
+  private ReplicationLogDiscoveryReplay discovery;
+
+  @Before
+  public void setUp() throws IOException {
+    conf = HBaseConfiguration.create();
+    localFs = FileSystem.getLocal(conf);
+    URI rootURI = new Path(testFolder.getRoot().getAbsolutePath()).toUri();
+    Path newFilesDirectory =
+      new Path(new Path(rootURI.getPath(), HA_GROUP_NAME), ReplicationLogReplay.IN_DIRECTORY_NAME);
+    ReplicationShardDirectoryManager shardManager =
+      new ReplicationShardDirectoryManager(conf, localFs, newFilesDirectory);
+    MetricsReplicationLogTracker metrics = new MetricsReplicationLogTrackerReplayImpl(HA_GROUP_NAME);
+    tracker = new TestableTracker(conf, HA_GROUP_NAME, shardManager, metrics);
+    tracker.init();
+    discovery = new ReplicationLogDiscoveryReplay(tracker);
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    if (tracker != null) {
+      tracker.close();
+    }
+    localFs.delete(new Path(testFolder.getRoot().toURI()), true);
+  }
+
+  /** Creates an empty in-progress log file with the given timestamp encoded in its name. */
+  private void createInProgressFile(long timestamp) throws IOException {
+    Path inProgressDir = tracker.getInProgressDirPath();
+    localFs.mkdirs(inProgressDir);
+    localFs.create(new Path(inProgressDir, timestamp + "_rs-1_uuid.plog"), true).close();
+  }
+
+  /**
+   * Ticket scenario: within round N a later file (T+30s) is moved to IN-PROGRESS while an older
+   * sibling (T+5s) is still waiting in the IN directory. The consistency point must align down to
+   * the round start (T), not the raw minimum IN-PROGRESS timestamp (T+30s).
+   */
+  @Test
+  public void testSyncStateAlignsMinInProgressTimestampToRoundStart() throws IOException {
+    // Only the later file (T+30s) is in IN-PROGRESS here. getConsistencyPoint() lists only the
+    // IN-PROGRESS directory, so an older sibling still waiting in the IN directory would not change
+    // the result; this test asserts purely that the min IN-PROGRESS timestamp is aligned down to
+    // the round start rather than used raw.
+    createInProgressFile(ROUND_START + 30000L);
+    ReplicationRound roundN = new ReplicationRound(ROUND_START, ROUND_START + ROUND_MILLIS);
+    discovery.setLastRoundInSync(roundN);
+    discovery.setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC);
+
+    assertEquals("Consistency point must align to round start, not the raw min in-progress ts",
+      ROUND_START, discovery.getConsistencyPoint());
+  }
+
+  /** A min IN-PROGRESS timestamp exactly on a round boundary is returned unchanged. */
+  @Test
+  public void testSyncStateMinOnRoundBoundaryReturnsBoundary() throws IOException {
+    createInProgressFile(ROUND_START);
+    ReplicationRound roundN = new ReplicationRound(ROUND_START, ROUND_START + ROUND_MILLIS);
+    discovery.setLastRoundInSync(roundN);
+    discovery.setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC);
+
+    assertEquals(ROUND_START, discovery.getConsistencyPoint());
+  }
+
+  /**
+   * When IN-PROGRESS files span two rounds (e.g. a retried file from round N-1 plus a round-N
+   * file), the minimum wins and is aligned to the earlier round's start.
+   */
+  @Test
+  public void testSyncStateMultipleRoundsUsesEarlierRoundStart() throws IOException {
+    long earlierRoundStart = ROUND_START - ROUND_MILLIS;
+    createInProgressFile(earlierRoundStart + 15000L); // round N-1 (retried file)
+    createInProgressFile(ROUND_START + 40000L);       // round N
+    ReplicationRound roundN = new ReplicationRound(ROUND_START, ROUND_START + ROUND_MILLIS);
+    discovery.setLastRoundInSync(roundN);
+    discovery.setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC);
+
+    assertEquals(earlierRoundStart, discovery.getConsistencyPoint());
+  }
+
+  /** Testable tracker exposing the protected in-progress directory path. */
+  private static class TestableTracker extends ReplicationLogTracker {
+    TestableTracker(Configuration conf, String haGroupName,
+      ReplicationShardDirectoryManager shardManager, MetricsReplicationLogTracker metrics) {
+      super(conf, haGroupName, shardManager, metrics);
+    }
+
+    @Override
+    public Path getInProgressDirPath() {
+      return super.getInProgressDirPath();
+    }
+  }
+}