This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature-new
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature-new by
this push:
new a746c5225a PHOENIX-7938 Round-align the sync-replication replay
consistency point (#2570)
a746c5225a is described below
commit a746c5225a69a13c1c8c48e2badb4c2137737b00
Author: Himanshu Gwalani <[email protected]>
AuthorDate: Thu Jul 16 20:12:58 2026 +0530
PHOENIX-7938 Round-align the sync-replication replay consistency point
(#2570)
---
.../reader/ReplicationLogDiscoveryReplay.java | 18 ++-
.../ReplicationLogDiscoveryReplayTestIT.java | 59 ++++++++
...tionLogDiscoveryReplayConsistencyPointTest.java | 150 +++++++++++++++++++++
3 files changed, 224 insertions(+), 3 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java
index 0b0f7ef517..2ef0b8abc1 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
@@ -555,7 +555,15 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
/**
* 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 class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
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()
+ .getNearestRoundStartTimestamp(minTimestampInProgress);
} 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.
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 6e3b1cc85e..7652db997b 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
@@ -653,6 +653,65 @@ public class ReplicationLogDiscoveryReplayTestIT extends
HABaseIT {
}
}
+ /**
+ * 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.
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayConsistencyPointTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayConsistencyPointTest.java
new file mode 100644
index 0000000000..33386f957a
--- /dev/null
+++
b/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();
+ }
+ }
+}
