You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by tk...@apache.org on 2026/07/29 16:45:55 UTC
(phoenix) branch PHOENIX-7562-feature-new updated: PHOENIX-7974 Preserve origin server identity when forwarding replication log files (#2588)
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 144322cf3b PHOENIX-7974 Preserve origin server identity when forwarding replication log files (#2588)
144322cf3b is described below
commit 144322cf3ba49651a4f4dfbbbd8468f7b0480974
Author: tkhurana <kh...@gmail.com>
AuthorDate: Wed Jul 29 09:45:49 2026 -0700
PHOENIX-7974 Preserve origin server identity when forwarding replication log files (#2588)
The forwarder computed the peer destination as getWriterPath(sourceTs,
forwardingServerName), keying the dst on the forwarding RS's own name and
discarding the origin writer identity. Two source files sharing a timestamp
(across RegionServers) then collapsed onto one dst; with FileUtil.copy
overwrite=false the second was permanently rejected with PathExistsException,
holding the HA group in SYNC_AND_FORWARD long after data was replicated.
Key the dst on the origin server name instead, so the forwarded file is
byte-identical to what the origin RS would have written natively and inherits
the writer's existing (server, timestamp) uniqueness invariant. Add
ReplicationLogTracker.getServerName() to extract the origin identity (dot-safe
so IP/FQDN host names survive), and route the .plog extension through a shared
ReplicationShardDirectoryManager.LOG_FILE_EXTENSION constant.
---
.../ReplicationLogDiscoveryForwarder.java | 6 ++-
.../phoenix/replication/ReplicationLogTracker.java | 36 ++++++++++++--
.../ReplicationShardDirectoryManager.java | 5 +-
.../phoenix/replication/tool/LogFileAnalyzer.java | 3 +-
.../ReplicationLogDiscoveryForwarderTest.java | 57 ++++++++++++++++++++++
.../replication/ReplicationLogTrackerTest.java | 34 +++++++++++++
6 files changed, 133 insertions(+), 8 deletions(-)
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
index c1a710362e..d9b9d1e911 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarder.java
@@ -98,8 +98,12 @@ public class ReplicationLogDiscoveryForwarder extends ReplicationLogDiscovery {
FileSystem srcFS = replicationLogTracker.getFileSystem();
FileStatus srcStat = srcFS.getFileStatus(src);
long ts = replicationLogTracker.getFileTimestamp(srcStat.getPath());
+ // Preserve the origin writer identity so the forwarded file is byte-identical to what the
+ // origin RS would have written natively. Re-deriving the name from (ts, forwardingServer) can
+ // collapse two distinct source files sharing a timestamp onto one dst, wedging SYNC re-entry.
+ String originServerName = replicationLogTracker.getServerName(srcStat.getPath());
ReplicationShardDirectoryManager remoteShardManager = logGroup.getOrCreatePeerShardManager();
- Path dst = remoteShardManager.getWriterPath(ts, logGroup.getServerName().getServerName());
+ Path dst = remoteShardManager.getWriterPath(ts, originServerName);
long startTime = EnvironmentEdgeManager.currentTimeMillis();
FileUtil.copy(srcFS, srcStat, remoteShardManager.getFileSystem(), dst, false, false, conf);
// successfully copied the file
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 b6a7a105e1..53395bb6d9 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
@@ -368,15 +368,13 @@ public class ReplicationLogTracker {
// Format: <ts>_<server>_<UUID>_<renameTs>.plog → extract prefix (first 2 parts)
String[] parts = fileName.split("_");
String prefix = parts[0] + "_" + parts[1];
- newFileName =
- prefix + "_" + UUID.randomUUID() + "_" + EnvironmentEdgeManager.currentTime() + ".plog";
+ newFileName = buildInProgressFileName(prefix);
targetDirectory = file.getParent();
} else {
// File is not in in-progress directory, add UUID + rename timestamp and move to
// IN_PROGRESS directory
String baseName = fileName.substring(0, fileName.lastIndexOf("."));
- newFileName =
- baseName + "_" + UUID.randomUUID() + "_" + EnvironmentEdgeManager.currentTime() + ".plog";
+ newFileName = buildInProgressFileName(baseName);
targetDirectory = getInProgressDirPath();
}
@@ -399,6 +397,18 @@ public class ReplicationLogTracker {
}
}
+ /**
+ * Builds an in-progress log file name from the given prefix by appending a fresh UUID, the
+ * current rename timestamp, and the log file extension. Format:
+ * {@code <prefix>_<UUID>_<renameTs>.plog}.
+ * @param prefix - the leading portion of the file name (e.g. {@code <ts>_<server>}).
+ * @return the in-progress file name
+ */
+ private String buildInProgressFileName(String prefix) {
+ return prefix + "_" + UUID.randomUUID() + "_" + EnvironmentEdgeManager.currentTime()
+ + ReplicationShardDirectoryManager.LOG_FILE_EXTENSION;
+ }
+
/**
* Validates if a file is a valid log file by checking if it ends with ".plog" extension.
* @param file - The file path to validate.
@@ -406,7 +416,7 @@ public class ReplicationLogTracker {
*/
protected boolean isValidLogFile(Path file) {
final String fileName = file.getName();
- return fileName.endsWith(".plog");
+ return fileName.endsWith(ReplicationShardDirectoryManager.LOG_FILE_EXTENSION);
}
/**
@@ -419,6 +429,22 @@ public class ReplicationLogTracker {
return Long.parseLong(parts[0]);
}
+ /**
+ * Extracts the server name (origin writer identity) from a log file name. Format:
+ * <ts>_<server>[_<UUID>_<renameTs>].plog → server is the second underscore-separated part. Server
+ * names ({@code host,port,startcode}) never contain underscores but do contain dots (e.g.
+ * {@code 10.244.2.10,16020,...}), so we strip only the trailing ".plog" suffix present on the
+ * 2-part (fresh) form and leave the 4-part (in-progress) form untouched.
+ * @param file - The file path to extract the server name from.
+ */
+ public String getServerName(Path file) {
+ String server = file.getName().split("_")[1];
+ return server.endsWith(ReplicationShardDirectoryManager.LOG_FILE_EXTENSION)
+ ? server.substring(0,
+ server.length() - ReplicationShardDirectoryManager.LOG_FILE_EXTENSION.length())
+ : server;
+ }
+
/**
* Extracts the UUID from an in-progress file name. Format: <ts>_<server>_<UUID>_<renameTs>.plog →
* UUID is the third underscore-separated part.
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 0b4edc10c4..174edc7ba9 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
@@ -64,11 +64,14 @@ public class ReplicationShardDirectoryManager {
*/
public static final String SHARD_DIR_FORMAT = "%03d";
+ /** File extension for replication log files. */
+ public static final String LOG_FILE_EXTENSION = ".plog";
+
/**
* Format string for log file names. <timestamp>_<servername>.plog Example
* 1762470665995_localhost,54575,1762470584502.plog
*/
- public static final String FILE_NAME_FORMAT = "%d_%s.plog";
+ public static final String FILE_NAME_FORMAT = "%d_%s" + LOG_FILE_EXTENSION;
/**
* Configuration key for the duration of each replication round in seconds.
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/tool/LogFileAnalyzer.java b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/tool/LogFileAnalyzer.java
index b61adc9fce..c3740d5d22 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/replication/tool/LogFileAnalyzer.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/replication/tool/LogFileAnalyzer.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
+import org.apache.phoenix.replication.ReplicationShardDirectoryManager;
import org.apache.phoenix.replication.log.LogFile;
import org.apache.phoenix.replication.log.LogFile.Record;
import org.apache.phoenix.replication.log.LogFileReader;
@@ -205,7 +206,7 @@ public class LogFileAnalyzer extends Configured implements Tool {
Path path = status.getPath();
if (status.isDirectory()) {
findLogFiles(path, files);
- } else if (path.getName().endsWith(".plog")) {
+ } else if (path.getName().endsWith(ReplicationShardDirectoryManager.LOG_FILE_EXTENSION)) {
files.add(path);
}
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java
index 65fb764a31..e646155a9f 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogDiscoveryForwarderTest.java
@@ -28,6 +28,9 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import java.io.IOException;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -36,6 +39,9 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
import org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode;
@@ -149,6 +155,57 @@ public class ReplicationLogDiscoveryForwarderTest extends ReplicationLogBaseTest
}
}
+ /**
+ * Two source files that share a timestamp but originate from different RegionServers must forward
+ * to two distinct destinations on the peer. The forwarder keys the dst on the origin writer
+ * identity; keying it on the forwarding RS instead would collapse both onto one dst and reject
+ * the second with a PathExistsException, wedging SYNC re-entry.
+ */
+ @Test
+ public void testForwardPreservesOriginServerIdentity() throws Exception {
+ ReplicationLogDiscoveryForwarder forwarder = logGroup.getLogForwarder();
+ ReplicationLogTracker localTracker = forwarder.getReplicationLogTracker();
+ ReplicationShardDirectoryManager localShardManager = logGroup.getLocalShardManager();
+ ReplicationShardDirectoryManager peerShardManager = logGroup.getOrCreatePeerShardManager();
+
+ // Same timestamp, different origin servers -- the collision scenario
+ long ts = EnvironmentEdgeManager.currentTimeMillis();
+ String originA = "10.244.1.10,16020,1784436416001";
+ String originB = "10.244.2.10,16020,1784436416002";
+
+ Path shardDir = localShardManager.getShardDirectory(ts);
+ localFs.mkdirs(shardDir);
+ Path srcA =
+ markInProgressSource(localTracker, new Path(shardDir, ts + "_" + originA + ".plog"));
+ Path srcB =
+ markInProgressSource(localTracker, new Path(shardDir, ts + "_" + originB + ".plog"));
+
+ forwarder.processFile(srcA);
+ forwarder.processFile(srcB);
+
+ // Both files must land on the peer under their own origin identity -- no collision.
+ Path peerShardDir = peerShardManager.getShardDirectory(ts);
+ FileSystem peerFs = peerShardManager.getFileSystem();
+ Set<String> peerNames = new HashSet<>();
+ for (FileStatus s : peerFs.listStatus(peerShardDir)) {
+ peerNames.add(s.getPath().getName());
+ }
+ assertEquals("Both origin files should be forwarded to distinct destinations", 2,
+ peerNames.size());
+ assertTrue("Peer should contain the file from originA",
+ peerNames.contains(ts + "_" + originA + ".plog"));
+ assertTrue("Peer should contain the file from originB",
+ peerNames.contains(ts + "_" + originB + ".plog"));
+ }
+
+ /** Creates an empty source file and moves it to the in-progress dir, as the forwarder expects. */
+ private Path markInProgressSource(ReplicationLogTracker tracker, Path file) throws IOException {
+ localFs.create(file, true).close();
+ Optional<Path> inProgress = tracker.markInProgress(file);
+ assertTrue("markInProgress should succeed", inProgress.isPresent());
+ return inProgress.get();
+ }
+
@Test
public void testSyncModeUpdateWaitTime() throws Exception {
final long[] waitTime = { 8L };
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java
index 6d6d8163a2..2204f05b88 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/replication/ReplicationLogTrackerTest.java
@@ -1157,6 +1157,40 @@ public class ReplicationLogTrackerTest {
oldFormatUUID.isPresent());
}
+ @Test
+ public void testGetServerName() throws IOException {
+ tracker.init();
+ // A simple server name, plus real HBase ServerNames (host,port,startcode) whose host part
+ // contains dots -- an IP-style k8s name and an FQDN. Both dotted forms must survive intact,
+ // since a naive dot-split would truncate them to the first label.
+ assertServerNameStableAcrossForms("rs1");
+ assertServerNameStableAcrossForms("10.244.2.10,16020,1784436416263");
+ assertServerNameStableAcrossForms("host1.example.com,16020,1784436416263");
+ }
+
+ /**
+ * Asserts that getServerName returns {@code serverName} for both the fresh 2-part file
+ * (<ts>_<server>.plog) and the 4-part in-progress form produced by the real markInProgress code
+ * (<ts>_<server>_<UUID>_<renameTs>.plog), rather than hand-crafting the in-progress name.
+ */
+ private void assertServerNameStableAcrossForms(String serverName) throws IOException {
+ ReplicationShardDirectoryManager shardManager = tracker.getReplicationShardDirectoryManager();
+ Path shardPath = shardManager.getAllShardPaths().get(0);
+ localFs.mkdirs(shardPath);
+
+ // Fresh 2-part file
+ Path freshFile = new Path(shardPath, "1704153600000_" + serverName + ".plog");
+ localFs.create(freshFile, true).close();
+ assertEquals("Server name should be extracted from fresh file", serverName,
+ tracker.getServerName(freshFile));
+
+ // Derive the 4-part in-progress name via the real code path instead of hand-crafting a UUID
+ Optional<Path> inProgressFile = tracker.markInProgress(freshFile);
+ assertTrue("markInProgress should succeed", inProgressFile.isPresent());
+ assertEquals("Server name should match on the in-progress form", serverName,
+ tracker.getServerName(inProgressFile.get()));
+ }
+
@Test
public void testGetOlderInProgressFiles() throws IOException {
// Initialize tracker