This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch PHOENIX-7562-feature
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/PHOENIX-7562-feature by this
push:
new 11a1f5a680 Phoenix-7672 Adding failover implementation via replay
(Addendum) (#2312)
11a1f5a680 is described below
commit 11a1f5a68005cc0247c9ba4b466b82b31e3700c5
Author: Himanshu Gwalani <[email protected]>
AuthorDate: Wed Nov 12 03:58:28 2025 +0530
Phoenix-7672 Adding failover implementation via replay (Addendum) (#2312)
---
.../replication/ReplicationLogDiscovery.java | 48 +-
.../phoenix/replication/ReplicationLogTracker.java | 2 +-
.../ReplicationShardDirectoryManager.java | 10 +-
.../reader/ReplicationLogDiscoveryReplay.java | 79 ++-
.../reader/ReplicationLogProcessor.java | 2 +
.../replication/reader/ReplicationLogReplay.java | 2 +
.../reader/ReplicationLogReplayService.java | 4 +
.../reader/RecoverLeaseFSUtilsTestIT.java} | 5 +-
.../ReplicationLogDiscoveryReplayTestIT.java} | 546 ++++++++++++++++++++-
.../reader/ReplicationLogProcessorTestIT.java} | 5 +-
.../reader/ReplicationLogReplayTestIT.java} | 5 +-
.../replication/ReplicationLogDiscoveryTest.java | 2 +-
.../ReplicationShardDirectoryManagerTest.java | 166 ++++++-
13 files changed, 796 insertions(+), 80 deletions(-)
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 76051edef4..01e9e13651 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
@@ -115,6 +115,7 @@ public abstract class ReplicationLogDiscovery {
}
public void init() throws IOException {
+ LOG.info("Initializing ReplicationLogDiscovery for haGroup: {}",
haGroupName);
initializeLastRoundProcessed();
this.metrics = createMetricsSource();
}
@@ -134,7 +135,7 @@ public abstract class ReplicationLogDiscovery {
public void start() throws IOException {
synchronized (this) {
if (isRunning) {
- LOG.warn("ReplicationLogDiscovery is already running for
group: {}", haGroupName);
+ LOG.warn("ReplicationLogDiscovery is already running for
haGroup: {}", haGroupName);
return;
}
// Initialize and schedule the executors
@@ -150,7 +151,7 @@ public abstract class ReplicationLogDiscovery {
}, 0, getReplayIntervalSeconds(), TimeUnit.SECONDS);
isRunning = true;
- LOG.info("ReplicationLogDiscovery started for group: {}",
haGroupName);
+ LOG.info("ReplicationLogDiscovery started for haGroup: {}",
haGroupName);
}
}
@@ -164,7 +165,7 @@ public abstract class ReplicationLogDiscovery {
synchronized (this) {
if (!isRunning) {
- LOG.warn("ReplicationLogDiscovery is not running for group:
{}", haGroupName);
+ LOG.warn("ReplicationLogDiscovery is not running for haGroup:
{}", haGroupName);
return;
}
@@ -185,7 +186,7 @@ public abstract class ReplicationLogDiscovery {
}
}
- LOG.info("ReplicationLogDiscovery stopped for group: {}", haGroupName);
+ LOG.info("ReplicationLogDiscovery stopped for haGroup: {}",
haGroupName);
}
/**
@@ -209,8 +210,8 @@ public abstract class ReplicationLogDiscovery {
try {
processRound(replicationRound);
} catch (IOException e) {
- LOG.error("Failed processing replication round {}. Will retry
in next "
- + "scheduled run.", replicationRound, e);
+ LOG.error("Failed processing replication round {} for haGroup
{}. Will retry"
+ + "in next scheduled run.", replicationRound,
haGroupName, e);
break; // stop this run, retry later
}
setLastRoundProcessed(replicationRound);
@@ -244,7 +245,7 @@ public abstract class ReplicationLogDiscovery {
* @throws IOException if there's an error during round processing
*/
protected void processRound(ReplicationRound replicationRound) throws
IOException {
- LOG.info("Starting to process round: {}", replicationRound);
+ LOG.info("Starting to process round: {} for haGroup: {}",
replicationRound, haGroupName);
// Increment the number of rounds processed
getMetrics().incrementNumRoundsProcessed();
@@ -254,7 +255,7 @@ public abstract class ReplicationLogDiscovery {
// Conditionally process the in progress files for the round
processInProgressDirectory();
}
- LOG.info("Finished processing round: {}", replicationRound);
+ LOG.info("Finished processing round: {} for haGroup: {}",
replicationRound, haGroupName);
}
/**
@@ -275,7 +276,8 @@ public abstract class ReplicationLogDiscovery {
* @throws IOException if there's an error during file processing
*/
protected void processNewFilesForRound(ReplicationRound replicationRound)
throws IOException {
- LOG.info("Starting new files processing for round: {}",
replicationRound);
+ LOG.info("Starting new files processing for round: {} for haGroup: {}",
+ replicationRound, haGroupName);
long startTime = EnvironmentEdgeManager.currentTime();
List<Path> files =
replicationLogTracker.getNewFilesForRound(replicationRound);
LOG.info("Number of new files for round {} is {}", replicationRound,
files.size());
@@ -284,7 +286,8 @@ public abstract class ReplicationLogDiscovery {
files =
replicationLogTracker.getNewFilesForRound(replicationRound);
}
long duration = EnvironmentEdgeManager.currentTime() - startTime;
- LOG.info("Finished new files processing for round: {} in {}ms",
replicationRound, duration);
+ LOG.info("Finished new files processing for round: {} in {}ms for
haGroup: {}",
+ replicationRound, duration, haGroupName);
getMetrics().updateTimeToProcessNewFiles(duration);
}
@@ -294,23 +297,25 @@ public abstract class ReplicationLogDiscovery {
* @throws IOException if there's an error during file processing
*/
protected void processInProgressDirectory() throws IOException {
+ LOG.info("Starting {} directory processing for haGroup: {}",
+ replicationLogTracker.getInProgressLogSubDirectoryName(),
haGroupName);
// Increase the count for number of times in progress directory is
processed
getMetrics().incrementNumInProgressDirectoryProcessed();
- LOG.info("Starting {} directory processing",
replicationLogTracker.getInProgressLogSubDirectoryName());
long startTime = EnvironmentEdgeManager.currentTime();
long oldestTimestampToProcess =
replicationLogTracker.getReplicationShardDirectoryManager()
.getNearestRoundStartTimestamp(EnvironmentEdgeManager.currentTime())
- getReplayIntervalSeconds() * 1000L;
List<Path> files =
replicationLogTracker.getOlderInProgressFiles(oldestTimestampToProcess);
- LOG.info("Number of {} files with oldestTimestampToProcess {} is {}",
+ LOG.info("Number of {} files with oldestTimestampToProcess {} is {}
for haGroup: {}",
replicationLogTracker.getInProgressLogSubDirectoryName(),
oldestTimestampToProcess,
- files.size());
+ files.size(), haGroupName);
while (!files.isEmpty()) {
processOneRandomFile(files);
files =
replicationLogTracker.getOlderInProgressFiles(oldestTimestampToProcess);
}
long duration = EnvironmentEdgeManager.currentTime() - startTime;
- LOG.info("Finished in-progress files processing in {}ms", duration);
+ LOG.info("Finished in-progress files processing in {}ms for haGroup:
{}",
+ duration, haGroupName);
getMetrics().updateTimeToProcessInProgressFiles(duration);
}
@@ -364,21 +369,26 @@ public abstract class ReplicationLogDiscovery {
Optional<Long> minTimestampFromInProgressFiles =
getMinTimestampFromInProgressFiles();
if (minTimestampFromInProgressFiles.isPresent()) {
- LOG.info("Initializing lastRoundProcessed from {} files with
minimum "
- + "timestamp as {}",
replicationLogTracker.getInProgressLogSubDirectoryName(),
minTimestampFromInProgressFiles.get());
+ LOG.info("Initializing lastRoundProcessed for haGroup: {} from {}
files with minimum "
+ + "timestamp as {}", haGroupName,
+ replicationLogTracker.getInProgressLogSubDirectoryName(),
+ minTimestampFromInProgressFiles.get());
this.lastRoundProcessed =
replicationLogTracker.getReplicationShardDirectoryManager()
.getReplicationRoundFromEndTime(minTimestampFromInProgressFiles.get());
} else {
Optional<Long> minTimestampFromNewFiles =
getMinTimestampFromNewFiles();
if (minTimestampFromNewFiles.isPresent()) {
- LOG.info("Initializing lastRoundProcessed from {} files with
minimum timestamp "
- + "as {}",
replicationLogTracker.getInSubDirectoryName(), minTimestampFromNewFiles.get());
+ LOG.info("Initializing lastRoundProcessed for haGroup: {} from
{}"
+ + "files with minimum timestamp as {}", haGroupName,
+ replicationLogTracker.getInSubDirectoryName(),
+ minTimestampFromNewFiles.get());
this.lastRoundProcessed = replicationLogTracker
.getReplicationShardDirectoryManager()
.getReplicationRoundFromEndTime(minTimestampFromNewFiles.get());
} else {
long currentTime = EnvironmentEdgeManager.currentTime();
- LOG.info("Initializing lastRoundProcessed from current time
{}", currentTime);
+ LOG.info("Initializing lastRoundProcessed for haGroup: {} from
current time {}",
+ haGroupName, currentTime);
this.lastRoundProcessed = replicationLogTracker
.getReplicationShardDirectoryManager()
.getReplicationRoundFromEndTime(EnvironmentEdgeManager.currentTime());
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java
index 6d16d9ed8a..5838a9de4b 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogTracker.java
@@ -116,7 +116,7 @@ public class ReplicationLogTracker {
* @return List of valid log file paths that belong to the specified
replication round
* @throws IOException if there's an error accessing the file system
*/
- protected List<Path> getNewFilesForRound(ReplicationRound
replicationRound) throws IOException {
+ public List<Path> getNewFilesForRound(ReplicationRound replicationRound)
throws IOException {
Path roundDirectory =
replicationShardDirectoryManager.getShardDirectory(replicationRound);
LOG.info("Getting new files for round {} from shard {}",
replicationRound, roundDirectory);
if (!fileSystem.exists(roundDirectory)) {
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationShardDirectoryManager.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationShardDirectoryManager.java
index e5fd495534..7010131b36 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationShardDirectoryManager.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationShardDirectoryManager.java
@@ -175,7 +175,7 @@ public class ReplicationShardDirectoryManager {
* @param timestamp The timestamp in milliseconds since epoch
* @return The nearest replication round start timestamp
*/
- protected long getNearestRoundStartTimestamp(long timestamp) {
+ public long getNearestRoundStartTimestamp(long timestamp) {
// Convert round time from seconds to milliseconds
long roundTimeMs = replicationRoundDurationSeconds * 1000L;
@@ -184,6 +184,14 @@ public class ReplicationShardDirectoryManager {
return (timestamp / roundTimeMs) * roundTimeMs;
}
+ public ReplicationRound getPreviousRound(final ReplicationRound
replicationRound) {
+ return getReplicationRoundFromEndTime(replicationRound.getStartTime());
+ }
+
+ public ReplicationRound getNextRound(final ReplicationRound
replicationRound) {
+ return getReplicationRoundFromStartTime(replicationRound.getEndTime());
+ }
+
public int getReplicationRoundDurationSeconds() {
return this.replicationRoundDurationSeconds;
}
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 93fd18c947..faf078fa7f 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
@@ -18,8 +18,6 @@
package org.apache.phoenix.replication.reader;
import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -136,6 +134,9 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
@Override
public void init() throws IOException {
+
+ LOG.info("Initializing ReplicationLogDiscoveryReplay for haGroup: {}",
haGroupName);
+
HAGroupStateListener degradedListener = (groupName, fromState,
toState, modifiedTime, clusterType, lastSyncStateTimeInMs) -> {
if (clusterType == ClusterType.LOCAL &&
HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(toState)) {
replicationReplayState.set(ReplicationReplayState.DEGRADED);
@@ -223,7 +224,10 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
*/
@Override
protected void initializeLastRoundProcessed() throws IOException {
+ LOG.info("Initializing last round processed for haGroup: {}",
haGroupName);
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()))
{
replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED,
ReplicationReplayState.DEGRADED);
@@ -298,6 +302,8 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
LOG.info("Starting replay with lastRoundProcessed={},
lastRoundInSync={}",
lastRoundProcessed, lastRoundInSync);
Optional<ReplicationRound> optionalNextRound =
getFirstRoundToProcess();
+ LOG.info("Found first round to process as {} for haGroup: {}",
+ optionalNextRound, haGroupName);
while (optionalNextRound.isPresent()) {
ReplicationRound replicationRound = optionalNextRound.get();
try {
@@ -314,9 +320,14 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
switch (currentState) {
case SYNCED_RECOVERY:
// Rewind to last in-sync round
- LOG.info("SYNCED_RECOVERY detected, rewinding to
lastRoundInSync={}",
+ LOG.info("SYNCED_RECOVERY detected, rewinding with
lastRoundInSync={}",
lastRoundInSync);
- setLastRoundProcessed(lastRoundInSync);
+ Optional<ReplicationRound> firstRoundToProcess =
getFirstRoundToProcess();
+ LOG.info("Calculated first round to process after
SYNCED_RECOVERY as"
+ + "{}", firstRoundToProcess);
+ firstRoundToProcess.ifPresent(round -> setLastRoundProcessed(
+
replicationLogTracker.getReplicationShardDirectoryManager()
+ .getPreviousRound(round)));
// Only reset to NORMAL if state hasn't been flipped to
DEGRADED
replicationReplayState.compareAndSet(ReplicationReplayState.SYNCED_RECOVERY,
ReplicationReplayState.SYNC);
@@ -348,17 +359,10 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
if (!optionalNextRound.isPresent() && shouldTriggerFailover()) {
LOG.info("No more rounds to process, lastRoundInSync={},
lastRoundProcessed={}. "
+ "Failover is triggered & in progress directory
is empty. "
- + "Marking cluster state as {}",
+ + "Attempting to mark cluster state as {}",
lastRoundInSync, lastRoundProcessed,
HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC);
- try {
- triggerFailover();
- LOG.info("Successfully updated the cluster state");
- failoverPending.set(false);
- } catch (InvalidClusterRoleTransitionException exception) {
- LOG.warn("Failed to update the cluster state.", exception);
- failoverPending.set(false);
- }
+ triggerFailover();
}
}
@@ -372,8 +376,15 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
* @return Optional containing the first round to process, or empty if not
enough time
* has passed
*/
- private Optional<ReplicationRound> getFirstRoundToProcess() {
- long lastRoundEndTimestamp = getLastRoundInSync().getEndTime();
+ private Optional<ReplicationRound> getFirstRoundToProcess() throws
IOException {
+ ReplicationRound lastRoundInSync = getLastRoundInSync();
+ long lastRoundEndTimestamp = lastRoundInSync.getEndTime();
+ if (lastRoundInSync.getStartTime() == 0) {
+ Optional<Long> optionalMinimumNewFilesTimestamp =
getMinTimestampFromNewFiles();
+ lastRoundEndTimestamp =
replicationLogTracker.getReplicationShardDirectoryManager()
+
.getNearestRoundStartTimestamp(optionalMinimumNewFilesTimestamp
+ .orElseGet(EnvironmentEdgeManager::currentTime));
+ }
long currentTime = EnvironmentEdgeManager.currentTime();
if (currentTime - lastRoundEndTimestamp < roundTimeMills +
bufferMillis) {
// nothing more to process
@@ -457,17 +468,39 @@ public class ReplicationLogDiscoveryReplay extends
ReplicationLogDiscovery {
return optionalHAGroupStateRecord.get();
}
+ /**
+ * Determines whether failover should be triggered based on completion
criteria.
+ *
+ * Failover is safe to trigger when all of the following conditions are
met:
+ * 1. A failover has been requested (failoverPending is true)
+ * 2. No files are currently in the in-progress directory
+ * 3. No new files exist for ongoing round
+ *
+ * These conditions ensure all replication logs have been processed before
transitioning
+ * the cluster from STANDBY to ACTIVE state.
+ *
+ * @return true if all conditions are met and failover should be
triggered, false otherwise
+ * @throws IOException if there's an error checking file status
+ */
protected boolean shouldTriggerFailover() throws IOException {
- return failoverPending.get() &&
lastRoundInSync.equals(lastRoundProcessed)
- && replicationLogTracker.getInProgressFiles().isEmpty()
- ;
- // TODO: Check for in files of lastRoundProcessed, lastRoundProcessed
+ roundTime as well - because there can be new files for upcoming round
+ return failoverPending.get() &&
replicationLogTracker.getInProgressFiles().isEmpty()
+ &&
replicationLogTracker.getNewFilesForRound(replicationLogTracker
+ .getReplicationShardDirectoryManager()
+ .getNextRound(getLastRoundProcessed())).isEmpty();
}
- protected void triggerFailover() throws IOException,
InvalidClusterRoleTransitionException {
- // TODO: Update cluster state to ACTIVE_IN_SYNC within try block
- // (once API is supported in HA Store)
- // Throw any exception back to caller.
+ protected void triggerFailover() {
+ try {
+
HAGroupStoreManager.getInstance(conf).setHAGroupStatusToSync(haGroupName);
+ failoverPending.set(false);
+ } catch (InvalidClusterRoleTransitionException
invalidClusterRoleTransitionException) {
+ LOG.warn("Failed to update the cluster state due to"
+ + "InvalidClusterRoleTransitionException. Setting
failoverPending"
+ + "to false.", invalidClusterRoleTransitionException);
+ failoverPending.set(false);
+ } catch (Exception exception) {
+ LOG.error("Failed to update the cluster state.", exception);
+ }
}
public enum ReplicationReplayState {
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
index 4a47779d70..e44bcc5ba9 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogProcessor.java
@@ -217,6 +217,8 @@ public class ReplicationLogProcessor implements Closeable {
public void processLogFile(FileSystem fs, Path filePath) throws
IOException {
+ LOG.info("Starting to process file {}", filePath);
+
// Map from Table Name to List of Mutations
Map<TableName, List<Mutation>> tableToMutationsMap = new HashMap<>();
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplay.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplay.java
index 5f12573403..587f9bdb02 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplay.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplay.java
@@ -107,6 +107,7 @@ public class ReplicationLogReplay {
* @throws IOException if there's an error during initialization
*/
protected void init() throws IOException {
+ LOG.info("Initializing ReplicationLogReplay for haGroup: {}",
haGroupName);
initializeFileSystem();
Path newFilesDirectory = new Path(new Path(rootURI.getPath(),
haGroupName), ReplicationLogReplay.IN_DIRECTORY_NAME);
ReplicationShardDirectoryManager replicationShardDirectoryManager =
@@ -121,6 +122,7 @@ public class ReplicationLogReplay {
}
public void close() {
+ LOG.info("Closing ReplicationLogReplay for haGroup: {}", haGroupName);
replicationLogDiscoveryReplay.getReplicationLogFileTracker().close();
replicationLogDiscoveryReplay.close();
// Remove the instance from cache
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplayService.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplayService.java
index 7df307f762..fef11daec6 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplayService.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogReplayService.java
@@ -195,6 +195,8 @@ public class ReplicationLogReplayService {
*/
protected void startReplicationReplay() throws IOException, SQLException {
List<String> replicationGroups = getReplicationGroups();
+ LOG.info("{} number of HA Groups found to start Replication Replay",
+ replicationGroups.size());
for (String replicationGroup : replicationGroups) {
ReplicationLogReplay.get(conf, replicationGroup).startReplay();
}
@@ -205,6 +207,8 @@ public class ReplicationLogReplayService {
*/
protected void stopReplicationReplay() throws IOException, SQLException {
List<String> replicationGroups = getReplicationGroups();
+ LOG.info("{} number of HA Groups found to stop Replication Replay",
+ replicationGroups.size());
for (String replicationGroup : replicationGroups) {
ReplicationLogReplay replicationLogReplay =
ReplicationLogReplay.get(conf, replicationGroup);
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/RecoverLeaseFSUtilsTest.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/RecoverLeaseFSUtilsTestIT.java
similarity index 96%
rename from
phoenix-core/src/test/java/org/apache/phoenix/replication/reader/RecoverLeaseFSUtilsTest.java
rename to
phoenix-core/src/it/java/org/apache/phoenix/replication/reader/RecoverLeaseFSUtilsTestIT.java
index 6df8577095..d688d55747 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/RecoverLeaseFSUtilsTest.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/RecoverLeaseFSUtilsTestIT.java
@@ -30,18 +30,21 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.util.CancelableProgressable;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
+import org.junit.experimental.categories.Category;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;
/**
* Test our recoverLease loop against mocked up filesystem.
*/
-public class RecoverLeaseFSUtilsTest extends ParallelStatsDisabledIT {
+@Category(NeedsOwnMiniClusterTest.class)
+public class RecoverLeaseFSUtilsTestIT extends ParallelStatsDisabledIT {
@ClassRule
public static TemporaryFolder testFolder = new TemporaryFolder();
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTest.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java
similarity index 72%
rename from
phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTest.java
rename to
phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java
index b0bb1d7e98..ca1171d76f 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTest.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplayTestIT.java
@@ -19,9 +19,11 @@ package org.apache.phoenix.replication.reader;
import java.io.IOException;
import java.net.URI;
+import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
@@ -31,6 +33,7 @@ import org.apache.hadoop.hbase.util.EnvironmentEdge;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.jdbc.ClusterRoleRecord;
+import org.apache.phoenix.jdbc.HAGroupStoreManager;
import org.apache.phoenix.jdbc.HAGroupStoreRecord;
import org.apache.phoenix.jdbc.HighAvailabilityPolicy;
import org.apache.phoenix.jdbc.HighAvailabilityTestingUtility;
@@ -65,9 +68,9 @@ import static org.junit.Assert.assertNotNull;
import static org.mockito.Mockito.when;
@Category(NeedsOwnMiniClusterTest.class)
-public class ReplicationLogDiscoveryReplayTest extends BaseTest {
+public class ReplicationLogDiscoveryReplayTestIT extends BaseTest {
- private static final Logger LOG =
LoggerFactory.getLogger(ReplicationLogDiscoveryReplayTest.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(ReplicationLogDiscoveryReplayTestIT.class);
private static final
HighAvailabilityTestingUtility.HBaseTestingUtilityPair CLUSTERS = new
HighAvailabilityTestingUtility.HBaseTestingUtilityPair();
private String zkUrl;
@@ -94,7 +97,7 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
public void setUp() throws Exception {
zkUrl = getLocalZkUrl(config);
peerZkUrl = CLUSTERS.getZkUrl2();
-
HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(testName.getMethodName(),
zkUrl, peerZkUrl,
+ HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(haGroupName,
zkUrl, peerZkUrl,
CLUSTERS.getMasterAddress1(), CLUSTERS.getMasterAddress2(),
ClusterRoleRecord.ClusterRole.ACTIVE,
ClusterRoleRecord.ClusterRole.STANDBY,
null);
@@ -1191,6 +1194,354 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
}
}
+ /**
+ * Tests replay method when lastRoundInSync is not set, i.e. has start
time 0.
+ * Validates that getFirstRoundToProcess uses minimum timestamp from new
files
+ */
+ @Test
+ public void testReplay_LastRoundInSync_NotInitialized() throws IOException
{
+ TestableReplicationLogTracker fileTracker =
createReplicationLogTracker(config, haGroupName, localFs, standbyUri);
+
+ try {
+ long roundTimeMills =
fileTracker.getReplicationShardDirectoryManager().getReplicationRoundDurationSeconds()
* 1000L;
+ long bufferMillis = (long) (roundTimeMills * 0.15);
+
+ // Create new files with specific timestamps
+ // The minimum timestamp should be used as the basis for the first
round
+ long minFileTimestamp = 1704067200000L + (5 * roundTimeMills) +
(30 * 1000L); // 2024-01-01 00:05:30
+ long maxFileTimestamp = 1704067200000L + (8 * roundTimeMills) +
(45 * 1000L); // 2024-01-01 00:08:45
+ long middleFileTimestamp = 1704067200000L + (7 * roundTimeMills) +
(15 * 1000L); // 2024-01-01 00:07:15
+
+ // Create files in different shard directories
+ ReplicationShardDirectoryManager shardManager =
fileTracker.getReplicationShardDirectoryManager();
+
+ // Create file with minimum timestamp
+ Path shardPath1 = shardManager.getShardDirectory(minFileTimestamp);
+ localFs.mkdirs(shardPath1);
+ Path file1 = new Path(shardPath1, minFileTimestamp + "_rs-1.plog");
+ localFs.create(file1, true).close();
+
+ // Create file with middle timestamp
+ Path shardPath2 =
shardManager.getShardDirectory(middleFileTimestamp);
+ localFs.mkdirs(shardPath2);
+ Path file2 = new Path(shardPath2, middleFileTimestamp +
"_rs-2.plog");
+ localFs.create(file2, true).close();
+
+ // Create file with maximum timestamp
+ Path shardPath3 = shardManager.getShardDirectory(maxFileTimestamp);
+ localFs.mkdirs(shardPath3);
+ Path file3 = new Path(shardPath3, maxFileTimestamp + "_rs-3.plog");
+ localFs.create(file3, true).close();
+
+ // Create HAGroupStoreRecord for STANDBY state
+ HAGroupStoreRecord mockRecord = new
HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, haGroupName,
+ HAGroupStoreRecord.HAGroupState.STANDBY, 0L,
HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl, zkUrl, peerZkUrl, 0L);
+
+ // Calculate expected first round start time (minimum timestamp
rounded down to nearest round start)
+ long expectedFirstRoundStart =
shardManager.getNearestRoundStartTimestamp(minFileTimestamp);
+ long expectedFirstRoundEnd = expectedFirstRoundStart +
roundTimeMills;
+
+ // Set current time to allow processing exactly 5 rounds
+ // Round 1 ends at expectedFirstRoundEnd
+ // Round 2 ends at expectedFirstRoundEnd + roundTimeMills
+ // Round 3 ends at expectedFirstRoundEnd + 2*roundTimeMills
+ // Round 4 ends at expectedFirstRoundEnd + 3*roundTimeMills
+ // Round 5 ends at expectedFirstRoundEnd + 4*roundTimeMills
+ // Round 6 starts at expectedFirstRoundEnd + 4*roundTimeMills
+ // For round 6 to be processable: currentTime -
(expectedFirstRoundEnd + 4*roundTimeMills) >= roundTimeMills + bufferMillis
+ // So: currentTime >= expectedFirstRoundEnd + 5*roundTimeMills +
bufferMillis
+ // For round 6 NOT to be processable: currentTime -
(expectedFirstRoundEnd + 5*roundTimeMills) < roundTimeMills + bufferMillis
+ // So exactly 5 rounds should be processed with currentTime =
expectedFirstRoundEnd + 5*roundTimeMills + bufferMillis - 1
+ // To be safe, use: currentTime = expectedFirstRoundEnd +
5*roundTimeMills + bufferMillis - 1000
+ long currentTime = expectedFirstRoundEnd + (5 * roundTimeMills) +
bufferMillis - 1000;
+ EnvironmentEdge edge = () -> currentTime;
+ EnvironmentEdgeManager.injectEdge(edge);
+
+ // Calculate exact number of rounds that should be processed
+ // Round 1: from getFirstRoundToProcess (based on lastRoundInSync
with start time 0)
+ // - Uses minFileTimestamp rounded down = expectedFirstRoundStart
+ // - Round 1: expectedFirstRoundStart to expectedFirstRoundEnd
+ // Round 2+: from getNextRoundToProcess (based on previous round's
end time)
+ // - Round 2: expectedFirstRoundEnd to expectedFirstRoundEnd +
roundTimeMills
+ // - Round 3: expectedFirstRoundEnd + roundTimeMills to
expectedFirstRoundEnd + 2*roundTimeMills
+ // - Round 4: expectedFirstRoundEnd + 2*roundTimeMills to
expectedFirstRoundEnd + 3*roundTimeMills
+ // - Round 5: expectedFirstRoundEnd + 3*roundTimeMills to
expectedFirstRoundEnd + 4*roundTimeMills
+ int expectedRoundCount = 5;
+ long[] expectedRoundStarts = new long[expectedRoundCount];
+ long[] expectedRoundEnds = new long[expectedRoundCount];
+ expectedRoundStarts[0] = expectedFirstRoundStart;
+ expectedRoundEnds[0] = expectedFirstRoundEnd;
+ for (int i = 1; i < expectedRoundCount; i++) {
+ expectedRoundStarts[i] = expectedFirstRoundEnd + ((i - 1) *
roundTimeMills);
+ expectedRoundEnds[i] = expectedRoundStarts[i] + roundTimeMills;
+ }
+
+ try {
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker,
mockRecord);
+
+ // Set lastRoundInSync with start time 0 (the new case being
tested)
+ ReplicationRound lastRoundInSyncWithZeroStart = new
ReplicationRound(0L, roundTimeMills);
+ discovery.setLastRoundInSync(lastRoundInSyncWithZeroStart);
+
+ // Set lastRoundProcessed to some initial value
+ ReplicationRound initialLastRoundProcessed = new
ReplicationRound(0L, roundTimeMills);
+ discovery.setLastRoundProcessed(initialLastRoundProcessed);
+
discovery.setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC);
+
+ // Verify initial state
+ ReplicationRound lastRoundInSyncBefore =
discovery.getLastRoundInSync();
+ assertEquals("lastRoundInSync should have start time 0 before
replay", 0L, lastRoundInSyncBefore.getStartTime());
+
+ // Call replay - should start from minimum timestamp of new
files
+ discovery.replay();
+
+ // Verify exact count of processRound calls
+ int actualProcessRoundCallCount =
discovery.getProcessRoundCallCount();
+ assertEquals("processRound should be called exactly " +
expectedRoundCount + " times",
+ expectedRoundCount, actualProcessRoundCallCount);
+
+ // Get all processed rounds
+ List<ReplicationRound> processedRounds =
discovery.getProcessedRounds();
+ assertEquals("Should have processed exactly " +
expectedRoundCount + " rounds",
+ expectedRoundCount, processedRounds.size());
+
+ // Verify each round's exact parameters
+ for (int i = 0; i < expectedRoundCount; i++) {
+ ReplicationRound actualRound = processedRounds.get(i);
+ assertEquals("Round " + (i + 1) + " should have correct
start time",
+ expectedRoundStarts[i],
actualRound.getStartTime());
+ assertEquals("Round " + (i + 1) + " should have correct
end time",
+ expectedRoundEnds[i], actualRound.getEndTime());
+ }
+
+ // Verify lastRoundProcessed was updated to the last processed
round
+ ReplicationRound lastRoundAfterReplay =
discovery.getLastRoundProcessed();
+ assertNotNull("Last round processed should not be null",
lastRoundAfterReplay);
+ assertEquals("Last round processed should match the last
processed round start time",
+ expectedRoundStarts[expectedRoundCount - 1],
lastRoundAfterReplay.getStartTime());
+ assertEquals("Last round processed should match the last
processed round end time",
+ expectedRoundEnds[expectedRoundCount - 1],
lastRoundAfterReplay.getEndTime());
+
+ // Verify lastRoundInSync was updated (in SYNC state, both
should advance together)
+ ReplicationRound lastRoundInSyncAfter =
discovery.getLastRoundInSync();
+ assertNotNull("Last round in sync should not be null",
lastRoundInSyncAfter);
+ assertEquals("Last round in sync should match last round
processed in SYNC state",
+ lastRoundAfterReplay, lastRoundInSyncAfter);
+ assertEquals("Last round in sync should have correct start
time",
+ expectedRoundStarts[expectedRoundCount - 1],
lastRoundInSyncAfter.getStartTime());
+ assertEquals("Last round in sync should have correct end time",
+ expectedRoundEnds[expectedRoundCount - 1],
lastRoundInSyncAfter.getEndTime());
+
+ // Verify state remains SYNC
+ assertEquals("State should remain SYNC",
+
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ } finally {
+ fileTracker.close();
+ }
+ }
+
+ /**
+ * Tests replay method when state changes from DEGRADED to SYNC between
round processing
+ * and lastRoundInSync has start time 0.
+ *
+ * Expected behavior:
+ * - 1st round is processed in DEGRADED state
+ * - 2nd round is processed in DEGRADED state, but then state transitions
to SYNCED_RECOVERY
+ * - When state becomes SYNCED_RECOVERY, lastRoundProcessed is rewound to
the previous round of getFirstRoundToProcess()
+ * (which uses minimum file timestamp when lastRoundInSync.startTime ==
0)
+ * - Then 5 rounds are processed starting from the first round (based on
minimum file timestamp) in SYNC state
+ *
+ * Total: 7 rounds processed (2 in DEGRADED, then rewind, then 5 in SYNC)
+ */
+ @Test
+ public void testReplay_DegradedToSync_LastRoundInSyncStartTimeZero()
throws IOException {
+ TestableReplicationLogTracker fileTracker =
createReplicationLogTracker(config, haGroupName, localFs, standbyUri);
+
+ try {
+ long roundTimeMills =
fileTracker.getReplicationShardDirectoryManager().getReplicationRoundDurationSeconds()
* 1000L;
+ long bufferMillis = (long) (roundTimeMills * 0.15);
+
+ // Create new files with specific timestamps
+ // The minimum timestamp should be used as the basis for the first
round
+ long minFileTimestamp = 1704067200000L + (5 * roundTimeMills) +
(30 * 1000L); // 2024-01-01 00:05:30
+ long maxFileTimestamp = 1704067200000L + (8 * roundTimeMills) +
(45 * 1000L); // 2024-01-01 00:08:45
+
+ // Create files in different shard directories
+ ReplicationShardDirectoryManager shardManager =
fileTracker.getReplicationShardDirectoryManager();
+
+ // Create file with minimum timestamp
+ Path shardPath1 = shardManager.getShardDirectory(minFileTimestamp);
+ localFs.mkdirs(shardPath1);
+ Path file1 = new Path(shardPath1, minFileTimestamp + "_rs-1.plog");
+ localFs.create(file1, true).close();
+
+ // Create file with maximum timestamp
+ Path shardPath2 = shardManager.getShardDirectory(maxFileTimestamp);
+ localFs.mkdirs(shardPath2);
+ Path file2 = new Path(shardPath2, maxFileTimestamp + "_rs-2.plog");
+ localFs.create(file2, true).close();
+
+ // Create HAGroupStoreRecord for STANDBY state
+ HAGroupStoreRecord mockRecord = new
HAGroupStoreRecord(HAGroupStoreRecord.DEFAULT_PROTOCOL_VERSION, haGroupName,
+ HAGroupStoreRecord.HAGroupState.STANDBY, 0L,
HighAvailabilityPolicy.FAILOVER.toString(), peerZkUrl, zkUrl, peerZkUrl, 0L);
+
+ // Calculate expected first round start time (minimum timestamp
rounded down to nearest round start)
+ long expectedFirstRoundStart =
shardManager.getNearestRoundStartTimestamp(minFileTimestamp);
+ long expectedFirstRoundEnd = expectedFirstRoundStart +
roundTimeMills;
+
+ // After rewind, lastRoundProcessed will be set to previous round
of firstRoundToProcess
+ // firstRoundToProcess = expectedFirstRoundStart to
expectedFirstRoundEnd
+ // previous round = expectedFirstRoundStart - roundTimeMills to
expectedFirstRoundStart
+ // Then getNextRoundToProcess will return the first round starting
from expectedFirstRoundStart
+ // So the first SYNC round after rewind will be
expectedFirstRoundStart to expectedFirstRoundEnd
+
+ // Set current time to allow processing 7 rounds total (2 in
DEGRADED, then rewind, then 5 in SYNC)
+ // After rewind, we need time for 5 more rounds starting from
expectedFirstRoundStart
+ // Round 7 (the 5th round after rewind) will end at
expectedFirstRoundEnd + 4*roundTimeMills
+ // = expectedFirstRoundStart + roundTimeMills + 4*roundTimeMills
= expectedFirstRoundStart + 5*roundTimeMills
+ // For round 7 to be processable: currentTime -
(expectedFirstRoundEnd + 4*roundTimeMills) >= roundTimeMills + bufferMillis
+ // So: currentTime >= expectedFirstRoundStart + 5*roundTimeMills +
bufferMillis
+ // Set it slightly above to ensure round 7 is processable
+ long currentTime = expectedFirstRoundStart + (5 * roundTimeMills)
+ bufferMillis + 1000;
+ EnvironmentEdge edge = () -> currentTime;
+ EnvironmentEdgeManager.injectEdge(edge);
+
+ // Calculate expected rounds
+ // Rounds 1-2: In DEGRADED state
+ // Round 1: expectedFirstRoundStart to expectedFirstRoundEnd
(DEGRADED)
+ // Round 2: expectedFirstRoundEnd to expectedFirstRoundEnd +
roundTimeMills (DEGRADED, transitions to SYNCED_RECOVERY at end)
+ // After round 2, state becomes SYNCED_RECOVERY and causes rewind
+ // - getFirstRoundToProcess returns: expectedFirstRoundStart to
expectedFirstRoundEnd
+ // - lastRoundProcessed is set to: expectedFirstRoundStart -
roundTimeMills to expectedFirstRoundStart
+ // Rounds 3-7: In SYNC state (starting from first round based on
minimum file timestamp)
+ // Round 3: expectedFirstRoundStart to expectedFirstRoundEnd
(SYNC, after rewind - first round)
+ // Round 4: expectedFirstRoundEnd to expectedFirstRoundEnd +
roundTimeMills (SYNC)
+ // Round 5: expectedFirstRoundEnd + roundTimeMills to
expectedFirstRoundEnd + 2*roundTimeMills (SYNC)
+ // Round 6: expectedFirstRoundEnd + 2*roundTimeMills to
expectedFirstRoundEnd + 3*roundTimeMills (SYNC)
+ // Round 7: expectedFirstRoundEnd + 3*roundTimeMills to
expectedFirstRoundEnd + 4*roundTimeMills (SYNC)
+ int expectedRoundCount = 7;
+ int degradedRoundCount = 2; // First 2 rounds are in DEGRADED state
+ int syncRoundCount = 5; // Last 5 rounds are in SYNC state (after
rewind)
+ long[] expectedRoundStarts = new long[expectedRoundCount];
+ long[] expectedRoundEnds = new long[expectedRoundCount];
+
+ // Rounds 1-2: DEGRADED
+ expectedRoundStarts[0] = expectedFirstRoundStart;
+ expectedRoundEnds[0] = expectedFirstRoundEnd;
+ expectedRoundStarts[1] = expectedFirstRoundEnd;
+ expectedRoundEnds[1] = expectedFirstRoundEnd + roundTimeMills;
+
+ // Rounds 3-7: SYNC (after rewind, starting from
expectedFirstRoundStart - the first round based on minimum file timestamp)
+ expectedRoundStarts[2] = expectedFirstRoundStart; // Round 3:
first round after rewind
+ expectedRoundEnds[2] = expectedFirstRoundEnd;
+ for (int i = 3; i < expectedRoundCount; i++) {
+ expectedRoundStarts[i] = expectedFirstRoundEnd + ((i - 3) *
roundTimeMills);
+ expectedRoundEnds[i] = expectedRoundStarts[i] + roundTimeMills;
+ }
+
+ try {
+ // Create discovery that transitions from DEGRADED to
SYNCED_RECOVERY after 2 rounds
+ // SYNCED_RECOVERY triggers a rewind, then transitions to SYNC
+ TestableReplicationLogDiscoveryReplay discovery =
+ new TestableReplicationLogDiscoveryReplay(fileTracker,
mockRecord) {
+ private int roundCount = 0;
+
+ @Override
+ protected void processRound(ReplicationRound
replicationRound) throws IOException {
+ super.processRound(replicationRound);
+ roundCount++;
+
+ // Transition to SYNCED_RECOVERY after 2 rounds (after
processing round 2)
+ // This triggers a rewind similar to when coming back
from DEGRADED to SYNC
+ if (roundCount == degradedRoundCount) {
+
setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNCED_RECOVERY);
+ }
+ }
+ };
+
+ // Set lastRoundInSync with start time 0
+ ReplicationRound lastRoundInSyncWithZeroStart = new
ReplicationRound(0L, roundTimeMills);
+ discovery.setLastRoundInSync(lastRoundInSyncWithZeroStart);
+
+ // Set lastRoundProcessed to some initial value
+ ReplicationRound initialLastRoundProcessed = new
ReplicationRound(0L, roundTimeMills);
+ discovery.setLastRoundProcessed(initialLastRoundProcessed);
+
discovery.setReplicationReplayState(ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED);
+
+ // Verify initial state
+ ReplicationRound lastRoundInSyncBefore =
discovery.getLastRoundInSync();
+ assertEquals("lastRoundInSync should have start time 0 before
replay", 0L, lastRoundInSyncBefore.getStartTime());
+ assertEquals("Initial state should be DEGRADED",
+
ReplicationLogDiscoveryReplay.ReplicationReplayState.DEGRADED,
+ discovery.getReplicationReplayState());
+
+ // Call replay - should start from minimum timestamp of new
files
+ discovery.replay();
+
+ // Verify exact count of processRound calls
+ int actualProcessRoundCallCount =
discovery.getProcessRoundCallCount();
+ assertEquals("processRound should be called exactly " +
expectedRoundCount + " times",
+ expectedRoundCount, actualProcessRoundCallCount);
+
+ // Get all processed rounds
+ List<ReplicationRound> processedRounds =
discovery.getProcessedRounds();
+ assertEquals("Should have processed exactly " +
expectedRoundCount + " rounds",
+ expectedRoundCount, processedRounds.size());
+
+ // Verify each round's exact parameters
+ for (int i = 0; i < expectedRoundCount; i++) {
+ ReplicationRound actualRound = processedRounds.get(i);
+ assertEquals("Round " + (i + 1) + " should have correct
start time",
+ expectedRoundStarts[i],
actualRound.getStartTime());
+ assertEquals("Round " + (i + 1) + " should have correct
end time",
+ expectedRoundEnds[i], actualRound.getEndTime());
+ }
+
+ // Verify lastRoundProcessed was updated to the last processed
round
+ ReplicationRound lastRoundAfterReplay =
discovery.getLastRoundProcessed();
+ assertNotNull("Last round processed should not be null",
lastRoundAfterReplay);
+ assertEquals("Last round processed should match the last
processed round start time",
+ expectedRoundStarts[expectedRoundCount - 1],
lastRoundAfterReplay.getStartTime());
+ assertEquals("Last round processed should match the last
processed round end time",
+ expectedRoundEnds[expectedRoundCount - 1],
lastRoundAfterReplay.getEndTime());
+
+ // Verify lastRoundInSync behavior:
+ // - During first 2 rounds (DEGRADED), lastRoundInSync should
NOT be updated (remains with start time 0)
+ // - After transition to SYNCED_RECOVERY and then SYNC (rounds
3-7), lastRoundInSync should be updated
+ // - Final lastRoundInSync should match lastRoundProcessed
(because we're in SYNC state at the end)
+ ReplicationRound lastRoundInSyncAfter =
discovery.getLastRoundInSync();
+ assertNotNull("Last round in sync should not be null",
lastRoundInSyncAfter);
+
+ // After processing, we're in SYNC state, so lastRoundInSync
should be updated
+ // It should match the last processed round because state is
SYNC
+ assertEquals("Last round in sync should match last round
processed in SYNC state",
+ lastRoundAfterReplay, lastRoundInSyncAfter);
+ assertEquals("Last round in sync should have correct start
time",
+ expectedRoundStarts[expectedRoundCount - 1],
lastRoundInSyncAfter.getStartTime());
+ assertEquals("Last round in sync should have correct end time",
+ expectedRoundEnds[expectedRoundCount - 1],
lastRoundInSyncAfter.getEndTime());
+ assertTrue("Last round in sync start time should not be 0
after processing in SYNC state",
+ lastRoundInSyncAfter.getStartTime() > 0);
+
+ // Verify state transitioned to SYNC
+ assertEquals("State should be SYNC after transition",
+
ReplicationLogDiscoveryReplay.ReplicationReplayState.SYNC,
+ discovery.getReplicationReplayState());
+
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ } finally {
+ fileTracker.close();
+ }
+ }
+
/**
* Tests replay method when failoverPending becomes true during processing
and triggers failover after all rounds.
* Validates that triggerFailover is called exactly once when all
conditions are met.
@@ -1299,7 +1650,7 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
/**
* Tests the shouldTriggerFailover method with various combinations of
failoverPending,
- * lastRoundInSync, lastRoundProcessed and in-progress files state.
+ * in-progress files, and new files for next round.
*/
@Test
public void testShouldTriggerFailover() throws IOException {
@@ -1317,11 +1668,12 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
try {
// Create test rounds
ReplicationRound testRound = new ReplicationRound(1704153600000L,
1704153660000L);
- ReplicationRound differentRound = new
ReplicationRound(1704153540000L, 1704153600000L);
+ ReplicationRound nextRound =
tracker.getReplicationShardDirectoryManager().getNextRound(testRound);
// Test Case 1: All conditions true - should return true
{
when(tracker.getInProgressFiles()).thenReturn(Collections.emptyList());
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.emptyList());
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
discovery.setLastRoundProcessed(testRound);
@@ -1334,6 +1686,7 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
// Test Case 2: failoverPending is false - should return false
{
when(tracker.getInProgressFiles()).thenReturn(Collections.emptyList());
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.emptyList());
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
discovery.setLastRoundProcessed(testRound);
@@ -1343,72 +1696,78 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
discovery.shouldTriggerFailover());
}
- // Test Case 3: lastRoundInSync not equals lastRoundProcessed -
should return false
+ // Test Case 3: in-progress files not empty - should return false
{
-
when(tracker.getInProgressFiles()).thenReturn(Collections.emptyList());
+
when(tracker.getInProgressFiles()).thenReturn(Collections.singletonList(new
Path("test.plog")));
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.emptyList());
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
- discovery.setLastRoundProcessed(differentRound);
+ discovery.setLastRoundProcessed(testRound);
discovery.setFailoverPending(true);
- assertFalse("Should not trigger failover when lastRoundInSync
!= lastRoundProcessed",
+ assertFalse("Should not trigger failover when in-progress
files are not empty",
discovery.shouldTriggerFailover());
}
- // Test Case 4: in-progress files not empty - should return false
+ // Test Case 4: new files exist for next round - should return
false
{
-
when(tracker.getInProgressFiles()).thenReturn(Collections.singletonList(new
Path("test.plog")));
+
when(tracker.getInProgressFiles()).thenReturn(Collections.emptyList());
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.singletonList(new
Path("test.plog")));
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
discovery.setLastRoundProcessed(testRound);
discovery.setFailoverPending(true);
- assertFalse("Should not trigger failover when in-progress
files are not empty",
+ assertFalse("Should not trigger failover when new files exist
for next round",
discovery.shouldTriggerFailover());
}
- // Test Case 5: failoverPending false AND lastRoundInSync !=
lastRoundProcessed - should return false
+ // Test Case 5: failoverPending false AND in-progress files not
empty - should return false
{
-
when(tracker.getInProgressFiles()).thenReturn(Collections.emptyList());
+
when(tracker.getInProgressFiles()).thenReturn(Collections.singletonList(new
Path("test.plog")));
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.emptyList());
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
- discovery.setLastRoundProcessed(differentRound);
+ discovery.setLastRoundProcessed(testRound);
discovery.setFailoverPending(false);
- assertFalse("Should not trigger failover when failoverPending
is false and rounds differ",
+ assertFalse("Should not trigger failover when failoverPending
is false and in-progress files exist",
discovery.shouldTriggerFailover());
}
- // Test Case 6: failoverPending false AND in-progress files not
empty - should return false
+ // Test Case 6: failoverPending false AND new files exist - should
return false
{
-
when(tracker.getInProgressFiles()).thenReturn(Collections.singletonList(new
Path("test.plog")));
+
when(tracker.getInProgressFiles()).thenReturn(Collections.emptyList());
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.singletonList(new
Path("test.plog")));
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
discovery.setLastRoundProcessed(testRound);
discovery.setFailoverPending(false);
- assertFalse("Should not trigger failover when failoverPending
is false and files exist",
+ assertFalse("Should not trigger failover when failoverPending
is false and new files exist",
discovery.shouldTriggerFailover());
}
- // Test Case 7: lastRoundInSync != lastRoundProcessed AND
in-progress files not empty - should return false
+ // Test Case 7: in-progress files not empty AND new files exist -
should return false
{
-
when(tracker.getInProgressFiles()).thenReturn(Collections.singletonList(new
Path("test.plog")));
+
when(tracker.getInProgressFiles()).thenReturn(Collections.singletonList(new
Path("test1.plog")));
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.singletonList(new
Path("test2.plog")));
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
- discovery.setLastRoundProcessed(differentRound);
+ discovery.setLastRoundProcessed(testRound);
discovery.setFailoverPending(true);
- assertFalse("Should not trigger failover when rounds differ
and files exist",
+ assertFalse("Should not trigger failover when both in-progress
and new files exist",
discovery.shouldTriggerFailover());
}
// Test Case 8: All conditions false - should return false
{
when(tracker.getInProgressFiles()).thenReturn(Collections.singletonList(new
Path("test.plog")));
+
when(tracker.getNewFilesForRound(nextRound)).thenReturn(Collections.singletonList(new
Path("test2.plog")));
TestableReplicationLogDiscoveryReplay discovery = new
TestableReplicationLogDiscoveryReplay(tracker, haGroupStoreRecord);
discovery.setLastRoundInSync(testRound);
- discovery.setLastRoundProcessed(differentRound);
+ discovery.setLastRoundProcessed(testRound);
discovery.setFailoverPending(false);
assertFalse("Should not trigger failover when all conditions
are false",
@@ -1420,6 +1779,141 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
}
}
+ /**
+ * Tests the triggerFailover method to verify it properly updates cluster
state
+ * and handles exceptions correctly.
+ */
+ @Test
+ public void testTriggerFailover() throws IOException, SQLException {
+ final String haGroupName = "testTriggerFailoverHAGroup";
+ // Set up HA group store record with STANDBY_TO_ACTIVE state for
successful transition
+ HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(haGroupName,
zkUrl, peerZkUrl,
+ CLUSTERS.getMasterAddress1(), CLUSTERS.getMasterAddress2(),
+ ClusterRoleRecord.ClusterRole.STANDBY_TO_ACTIVE,
ClusterRoleRecord.ClusterRole.ACTIVE_TO_STANDBY,
+ null);
+
+ TestableReplicationLogTracker fileTracker = null;
+ try {
+ fileTracker = createReplicationLogTracker(config, haGroupName,
localFs, standbyUri);
+
+ // Create a spy of ReplicationLogDiscoveryReplay to test actual
implementation
+ ReplicationLogDiscoveryReplay discovery = Mockito.spy(
+ new ReplicationLogDiscoveryReplay(fileTracker));
+
+ // Initialize the discovery
+ discovery.init();
+
+ // Set failoverPending to true
+ discovery.setFailoverPending(true);
+ assertTrue("failoverPending should be true before
triggerFailover", discovery.getFailoverPending());
+
+ // Verify initial state is STANDBY_TO_ACTIVE
+ Optional<HAGroupStoreRecord> recordBefore =
HAGroupStoreManager.getInstance(config)
+ .getHAGroupStoreRecord(haGroupName);
+ assertTrue("HA group record should exist",
recordBefore.isPresent());
+ assertEquals("Initial state should be STANDBY_TO_ACTIVE",
+ HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE,
+ recordBefore.get().getHAGroupState());
+
+ // Call triggerFailover - should succeed and update cluster state
to ACTIVE_IN_SYNC
+ discovery.triggerFailover();
+
+ // Verify triggerFailover was called once
+ Mockito.verify(discovery, Mockito.times(1)).triggerFailover();
+
+ // Verify cluster state is updated to ACTIVE_IN_SYNC after failover
+ Optional<HAGroupStoreRecord> recordAfter =
HAGroupStoreManager.getInstance(config)
+ .getHAGroupStoreRecord(haGroupName);
+ assertTrue("HA group record should exist after failover",
recordAfter.isPresent());
+ assertEquals("Cluster state should be ACTIVE_IN_SYNC after
successful failover",
+ HAGroupStoreRecord.HAGroupState.ACTIVE_IN_SYNC,
+ recordAfter.get().getHAGroupState());
+
+ // Verify failoverPending is set to false after successful failover
+ assertFalse("failoverPending should be set to false after
successful triggerFailover",
+ discovery.getFailoverPending());
+
+ } finally {
+ // Clean up
+ if (fileTracker != null) {
+ fileTracker.close();
+ }
+ try {
+
HAGroupStoreTestUtil.deleteHAGroupRecordInSystemTable(haGroupName, zkUrl);
+ } catch (Exception e) {
+ LOG.warn("Failed to clean up HA group store record", e);
+ }
+ }
+ }
+
+ /**
+ * Tests triggerFailover when
HAGroupStoreManager.getInstance(conf).setHAGroupStatusToSync(haGroupName)
+ * throws InvalidClusterRoleTransitionException.
+ * Verifies that failoverPending is set to false even when the exception
occurs.
+ *
+ * This test sets up the HA group in STANDBY state instead of
STANDBY_TO_ACTIVE.
+ * Calling setHAGroupStatusToSync from STANDBY state will throw
InvalidClusterRoleTransitionException
+ * because the transition from STANDBY directly to ACTIVE_IN_SYNC is not
allowed.
+ */
+ @Test
+ public void
testTriggerFailover_InvalidClusterRoleTransitionExceptionFromHAGroupStoreManager()
throws IOException, SQLException {
+ final String haGroupName =
"testTriggerFailoverInvalidTransitionHAGroup";
+
+ // Set up HA group store record in STANDBY state (not
STANDBY_TO_ACTIVE)
+ // This will cause setHAGroupStatusToSync to throw
InvalidClusterRoleTransitionException
+ // because transitioning from STANDBY directly to ACTIVE_IN_SYNC is
invalid
+ HAGroupStoreTestUtil.upsertHAGroupRecordInSystemTable(haGroupName,
zkUrl, peerZkUrl,
+ CLUSTERS.getMasterAddress1(), CLUSTERS.getMasterAddress2(),
+ ClusterRoleRecord.ClusterRole.STANDBY,
ClusterRoleRecord.ClusterRole.ACTIVE,
+ null);
+
+ TestableReplicationLogTracker fileTracker = null;
+ try {
+ fileTracker = createReplicationLogTracker(config, haGroupName,
localFs, standbyUri);
+
+ // Create a spy of ReplicationLogDiscoveryReplay to test actual
implementation
+ ReplicationLogDiscoveryReplay discovery = Mockito.spy(
+ new ReplicationLogDiscoveryReplay(fileTracker));
+
+ // Initialize the discovery
+ discovery.init();
+
+ // Set failoverPending to true
+ discovery.setFailoverPending(true);
+ assertTrue("failoverPending should be true before
triggerFailover", discovery.getFailoverPending());
+
+ // Verify initial state is STANDBY (not STANDBY_TO_ACTIVE)
+ Optional<HAGroupStoreRecord> recordBefore =
HAGroupStoreManager.getInstance(config)
+ .getHAGroupStoreRecord(haGroupName);
+ assertTrue("HA group record should exist",
recordBefore.isPresent());
+ assertEquals("Initial state should be DEGRADED_STANDBY",
+ HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY,
+ recordBefore.get().getHAGroupState());
+
+ // Call triggerFailover - should handle
InvalidClusterRoleTransitionException
+ // because transitioning from DEGRADED_STANDBY directly to
ACTIVE_IN_SYNC is invalid
+ discovery.triggerFailover();
+
+ // Verify triggerFailover was called (implicitly through spy)
+ Mockito.verify(discovery, Mockito.times(1)).triggerFailover();
+
+ // Verify failoverPending is set to false even when
InvalidClusterRoleTransitionException occurs
+ assertFalse("failoverPending should be set to false when
InvalidClusterRoleTransitionException occurs",
+ discovery.getFailoverPending());
+
+ } finally {
+ // Clean up
+ if (fileTracker != null) {
+ fileTracker.close();
+ }
+ try {
+
HAGroupStoreTestUtil.deleteHAGroupRecordInSystemTable(haGroupName, zkUrl);
+ } catch (Exception e) {
+ LOG.warn("Failed to clean up HA group store record", e);
+ }
+ }
+ }
+
/**
* Testable implementation of ReplicationLogDiscoveryReplay for unit
testing.
* Provides dependency injection for HAGroupStoreRecord, tracks processed
rounds,
@@ -1482,9 +1976,7 @@ public class ReplicationLogDiscoveryReplayTest extends
BaseTest {
protected void triggerFailover() {
// Track calls to triggerFailover for validation
triggerFailoverCallCount++;
-
- // Simulate the real behavior: set failoverPending to false after
triggering failover
- setFailoverPending(false);
+ super.triggerFailover();
}
public int getTriggerFailoverCallCount() {
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTest.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
similarity index 99%
rename from
phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTest.java
rename to
phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
index 9540a8d20f..1ad72806f5 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTest.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogProcessorTestIT.java
@@ -66,6 +66,7 @@ import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryServices;
@@ -83,11 +84,13 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
+import org.junit.experimental.categories.Category;
import org.junit.rules.TemporaryFolder;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
-public class ReplicationLogProcessorTest extends ParallelStatsDisabledIT {
+@Category(NeedsOwnMiniClusterTest.class)
+public class ReplicationLogProcessorTestIT extends ParallelStatsDisabledIT {
private static final String CREATE_TABLE_SQL_STATEMENT = "CREATE TABLE %s
(ID VARCHAR PRIMARY KEY, " +
"COL_1 VARCHAR, COL_2 VARCHAR, COL_3 BIGINT)";
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogReplayTest.java
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogReplayTestIT.java
similarity index 97%
rename from
phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogReplayTest.java
rename to
phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogReplayTestIT.java
index 3412966abd..37c79b8852 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/reader/ReplicationLogReplayTest.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/replication/reader/ReplicationLogReplayTestIT.java
@@ -34,6 +34,7 @@ import java.util.Map;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.jdbc.ClusterRoleRecord;
import org.apache.phoenix.jdbc.HAGroupStoreClient;
import org.apache.phoenix.jdbc.HighAvailabilityTestingUtility;
@@ -43,10 +44,12 @@ import
org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
import org.apache.phoenix.util.HAGroupStoreTestUtil;
import org.apache.phoenix.util.ReadOnlyProps;
import org.junit.*;
+import org.junit.experimental.categories.Category;
import org.junit.rules.TemporaryFolder;
import org.junit.rules.TestName;
-public class ReplicationLogReplayTest extends BaseTest {
+@Category(NeedsOwnMiniClusterTest.class)
+public class ReplicationLogReplayTestIT extends BaseTest {
@ClassRule
public static TemporaryFolder testFolder = new TemporaryFolder();
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java
index f609a4aed2..0649b551b7 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryTest.java
@@ -436,7 +436,7 @@ public class ReplicationLogDiscoveryTest {
@Test
public void testShouldProcessInProgressDirectory() {
// Test multiple times to verify probability-based behavior
- int totalTests = 1000;
+ int totalTests = 100000;
int trueCount = 0;
for (int i = 0; i < totalTests; i++) {
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationShardDirectoryManagerTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationShardDirectoryManagerTest.java
index 0b413104fc..f26e8436e6 100644
---
a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationShardDirectoryManagerTest.java
+++
b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationShardDirectoryManagerTest.java
@@ -5,7 +5,7 @@
* 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
+ * with the License. You may obtain a copy of the License a
*
* http://www.apache.org/licenses/LICENSE-2.0
*
@@ -234,7 +234,7 @@ public class ReplicationShardDirectoryManagerTest {
// Test 7: Multiple rounds later
long multipleRoundsLater = dayStart + (300 * 1000L); // 00:05:00 (5
minutes)
- long expectedRoundStart = dayStart + (300 * 1000L); // Should be exact
+ long expectedRoundStart = dayStart + (300 * 1000L); // Should be exac
long result7 =
manager.getNearestRoundStartTimestamp(multipleRoundsLater);
assertEquals("Multiple rounds later should return exact round start",
expectedRoundStart, result7);
@@ -389,7 +389,7 @@ public class ReplicationShardDirectoryManagerTest {
@Test
public void testGetReplicationRoundFromStartTimeConsistency() {
- // Test that the same input always produces the same output
+ // Test that the same input always produces the same outpu
long timestamp = 1704110400000L; // 2024-01-01 12:00:00 UTC
ReplicationRound result1 =
manager.getReplicationRoundFromStartTime(timestamp);
@@ -543,7 +543,7 @@ public class ReplicationShardDirectoryManagerTest {
@Test
public void testGetReplicationRoundFromEndTimeConsistency() {
- // Test that the same input always produces the same output
+ // Test that the same input always produces the same outpu
long timestamp = 1704110400000L; // 2024-01-01 12:00:00 UTC
ReplicationRound result1 =
manager.getReplicationRoundFromEndTime(timestamp);
@@ -595,5 +595,161 @@ public class ReplicationShardDirectoryManagerTest {
assertEquals("Shard " + i + " should have correct name",
expectedShardName, shardPath.getName());
}
}
-}
+ @Test
+ public void testGetPreviousRound() {
+ // Use a specific day for consistent testing
+ // 2024-01-01 00:00:00 UTC = 1704067200000L
+ long dayStart = 1704067200000L; // 2024-01-01 00:00:00 UTC
+
+ // Default configuration: 60-second rounds
+ long roundDurationMs = 60 * 1000L; // 60 seconds in milliseconds
+
+ // Test 1: First round (00:00:00 to 00:01:00)
+ long firstRoundStart = dayStart; // 00:00:00
+ long firstRoundEnd = firstRoundStart + roundDurationMs; // 00:01:00
+ ReplicationRound firstRound = new ReplicationRound(firstRoundStart,
firstRoundEnd);
+ ReplicationRound previousRound = manager.getPreviousRound(firstRound);
+
+ // Previous round should end at firstRoundStart (00:00:00), which
rounds down to 00:00:00
+ // Start time should be end time - round duration = 00:00:00 - 60s =
-60s (edge case)
+ long expectedEnd = dayStart; // 00:00:00
+ long expectedStart = expectedEnd - roundDurationMs; // -60 seconds
+ assertEquals("Previous round end time should be start time of current
round rounded down",
+ expectedEnd, previousRound.getEndTime());
+ assertEquals("Previous round start time should be end time - round
duration",
+ expectedStart, previousRound.getStartTime());
+
+ // Test 2: Second round (00:01:00 to 00:02:00)
+ long secondRoundStart = dayStart + roundDurationMs; // 00:01:00
+ long secondRoundEnd = secondRoundStart + roundDurationMs; // 00:02:00
+ ReplicationRound secondRound = new ReplicationRound(secondRoundStart,
secondRoundEnd);
+ ReplicationRound previousRound2 =
manager.getPreviousRound(secondRound);
+
+ // Previous round should end at secondRoundStart (00:01:00), which
rounds down to 00:01:00
+ // Start time should be 00:01:00 - 60s = 00:00:00
+ long expectedEnd2 = secondRoundStart; // 00:01:00
+ long expectedStart2 = expectedEnd2 - roundDurationMs; // 00:00:00
+ assertEquals("Previous round end time should be start time of current
round rounded down",
+ expectedEnd2, previousRound2.getEndTime());
+ assertEquals("Previous round start time should be end time - round
duration",
+ expectedStart2, previousRound2.getStartTime());
+
+ // Test 3: Round with mid-timestamp start (00:01:30 would round to
00:01:00)
+ long midRoundStart = dayStart + (90 * 1000L); // 00:01:30 (not aligned
to round boundary)
+ long midRoundEnd = midRoundStart + roundDurationMs; // 00:02:30
+ ReplicationRound midRound = new ReplicationRound(midRoundStart,
midRoundEnd);
+ ReplicationRound previousRound3 = manager.getPreviousRound(midRound);
+
+ // Previous round should end at midRoundStart rounded down = 00:01:00
+ // Start time should be 00:01:00 - 60s = 00:00:00
+ long expectedEnd3 = dayStart + roundDurationMs; // 00:01:00 (rounded
down from 00:01:30)
+ long expectedStart3 = expectedEnd3 - roundDurationMs; // 00:00:00
+ assertEquals("Previous round end time should round down start time of
current round",
+ expectedEnd3, previousRound3.getEndTime());
+ assertEquals("Previous round start time should be end time - round
duration",
+ expectedStart3, previousRound3.getStartTime());
+
+ // Test 4: Multiple rounds later (00:05:00 to 00:06:00)
+ long multipleRoundsStart = dayStart + (5 * roundDurationMs); //
00:05:00
+ long multipleRoundsEnd = multipleRoundsStart + roundDurationMs; //
00:06:00
+ ReplicationRound multipleRounds = new
ReplicationRound(multipleRoundsStart, multipleRoundsEnd);
+ ReplicationRound previousRound4 =
manager.getPreviousRound(multipleRounds);
+
+ // Previous round should end at multipleRoundsStart = 00:05:00
+ // Start time should be 00:05:00 - 60s = 00:04:00
+ long expectedEnd4 = multipleRoundsStart; // 00:05:00
+ long expectedStart4 = expectedEnd4 - roundDurationMs; // 00:04:00
+ assertEquals("Previous round end time should be start time of current
round",
+ expectedEnd4, previousRound4.getEndTime());
+ assertEquals("Previous round start time should be end time - round
duration",
+ expectedStart4, previousRound4.getStartTime());
+
+ // Test 5: Verify round duration is consistent
+ long previousRoundDuration = previousRound4.getEndTime() -
previousRound4.getStartTime();
+ assertEquals("Previous round duration should be 60 seconds",
roundDurationMs, previousRoundDuration);
+ }
+
+ @Test
+ public void testGetNextRound() {
+ // Use a specific day for consistent testing
+ // 2024-01-01 00:00:00 UTC = 1704067200000L
+ long dayStart = 1704067200000L; // 2024-01-01 00:00:00 UTC
+
+ // Default configuration: 60-second rounds
+ long roundDurationMs = 60 * 1000L; // 60 seconds in milliseconds
+
+ // Test 1: First round (00:00:00 to 00:01:00)
+ long firstRoundStart = dayStart; // 00:00:00
+ long firstRoundEnd = firstRoundStart + roundDurationMs; // 00:01:00
+ ReplicationRound firstRound = new ReplicationRound(firstRoundStart,
firstRoundEnd);
+ ReplicationRound nextRound = manager.getNextRound(firstRound);
+
+ // Next round should start at firstRoundEnd (00:01:00), which rounds
down to 00:01:00
+ // End time should be 00:01:00 + 60s = 00:02:00
+ long expectedStart = firstRoundEnd; // 00:01:00
+ long expectedEnd = expectedStart + roundDurationMs; // 00:02:00
+ assertEquals("Next round start time should be end time of current
round rounded down",
+ expectedStart, nextRound.getStartTime());
+ assertEquals("Next round end time should be start time + round
duration",
+ expectedEnd, nextRound.getEndTime());
+
+ // Test 2: Second round (00:01:00 to 00:02:00)
+ long secondRoundStart = dayStart + roundDurationMs; // 00:01:00
+ long secondRoundEnd = secondRoundStart + roundDurationMs; // 00:02:00
+ ReplicationRound secondRound = new ReplicationRound(secondRoundStart,
secondRoundEnd);
+ ReplicationRound nextRound2 = manager.getNextRound(secondRound);
+
+ // Next round should start at secondRoundEnd (00:02:00), which rounds
down to 00:02:00
+ // End time should be 00:02:00 + 60s = 00:03:00
+ long expectedStart2 = secondRoundEnd; // 00:02:00
+ long expectedEnd2 = expectedStart2 + roundDurationMs; // 00:03:00
+ assertEquals("Next round start time should be end time of current
round rounded down",
+ expectedStart2, nextRound2.getStartTime());
+ assertEquals("Next round end time should be start time + round
duration",
+ expectedEnd2, nextRound2.getEndTime());
+
+ // Test 3: Round with mid-timestamp end (00:02:30 would round to
00:02:00)
+ long midRoundStart = dayStart + (120 * 1000L); // 00:02:00
+ long midRoundEnd = dayStart + (150 * 1000L); // 00:02:30 (not aligned
to round boundary)
+ ReplicationRound midRound = new ReplicationRound(midRoundStart,
midRoundEnd);
+ ReplicationRound nextRound3 = manager.getNextRound(midRound);
+
+ // Next round should start at midRoundEnd rounded down = 00:02:00
+ // End time should be 00:02:00 + 60s = 00:03:00
+ long expectedStart3 = dayStart + (120 * 1000L); // 00:02:00 (rounded
down from 00:02:30)
+ long expectedEnd3 = expectedStart3 + roundDurationMs; // 00:03:00
+ assertEquals("Next round start time should round down end time of
current round",
+ expectedStart3, nextRound3.getStartTime());
+ assertEquals("Next round end time should be start time + round
duration",
+ expectedEnd3, nextRound3.getEndTime());
+
+ // Test 4: Multiple rounds later (00:05:00 to 00:06:00)
+ long multipleRoundsStart = dayStart + (5 * roundDurationMs); //
00:05:00
+ long multipleRoundsEnd = multipleRoundsStart + roundDurationMs; //
00:06:00
+ ReplicationRound multipleRounds = new
ReplicationRound(multipleRoundsStart, multipleRoundsEnd);
+ ReplicationRound nextRound4 = manager.getNextRound(multipleRounds);
+
+ // Next round should start at multipleRoundsEnd = 00:06:00
+ // End time should be 00:06:00 + 60s = 00:07:00
+ long expectedStart4 = multipleRoundsEnd; // 00:06:00
+ long expectedEnd4 = expectedStart4 + roundDurationMs; // 00:07:00
+ assertEquals("Next round start time should be end time of current
round",
+ expectedStart4, nextRound4.getStartTime());
+ assertEquals("Next round end time should be start time + round
duration",
+ expectedEnd4, nextRound4.getEndTime());
+
+ // Test 5: Verify round duration is consistent
+ long nextRoundDuration = nextRound4.getEndTime() -
nextRound4.getStartTime();
+ assertEquals("Next round duration should be 60 seconds",
roundDurationMs, nextRoundDuration);
+
+ // Test 6: Verify continuity - next round of previous round should
equal original round
+ ReplicationRound originalRound = new ReplicationRound(dayStart + (3 *
roundDurationMs), dayStart + (4 * roundDurationMs)); // 00:03:00 to 00:04:00
+ ReplicationRound prevRound = manager.getPreviousRound(originalRound);
+ ReplicationRound nextOfPrev = manager.getNextRound(prevRound);
+ assertEquals("Next round of previous round should equal original round
start time",
+ originalRound.getStartTime(), nextOfPrev.getStartTime());
+ assertEquals("Next round of previous round should equal original round
end time",
+ originalRound.getEndTime(), nextOfPrev.getEndTime());
+ }
+}
