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 e3ac3900fd PHOENIX-7977 Stage forwarded replication log files in a
per-shard staging subdirectory before atomic rename (#2590)
e3ac3900fd is described below
commit e3ac3900fd0042bc17ba6c0ae684fbe7002bc6da
Author: tkhurana <[email protected]>
AuthorDate: Thu Jul 30 06:26:36 2026 -0700
PHOENIX-7977 Stage forwarded replication log files in a per-shard staging
subdirectory before atomic rename (#2590)
The forwarder copied a fallback-cluster log file straight onto its final
<ts>_<origin>.plog name on the peer via FileUtil.copy (overwrite=false).
Because FileUtil.copy publishes the file at that replay-eligible name before
close() completes, the standby replay consumer could pick up the
half-written
file, force-recover the HDFS lease, and cause the forwarder's close() to
fail
with LeaseExpiredException. That threw out of processFile before the
STORE_AND_FORWARD -> SYNC_AND_FORWARD throughput check, delaying the mode
transition.
Fix: copy into a per-shard .staging subdirectory
(<shard>/.staging/<ts>_<origin>.plog,
overwrite=true to reclaim orphans) and atomically rename it up to the final
.plog
in the shard directory once fully written. The staged file keeps its real
.plog
name but is invisible to replay because every listing gates on
FileStatus.isFile(), which skips the subdirectory and anything inside it.
Replay
only ever sees the atomically published file. Delivery remains at-least-once
(unchanged); replay idempotency makes a re-publish safe.
Tests: forwarder staging/rename/retry/orphan cases in
ReplicationLogDiscoveryForwarderTest; .staging-subdirectory exclusion
across the
listing entry points in ReplicationLogTrackerTest.
---
.../ReplicationLogDiscoveryForwarder.java | 39 ++++-
.../phoenix/replication/ReplicationLogTracker.java | 3 +-
.../ReplicationShardDirectoryManager.java | 27 +++
.../ReplicationLogDiscoveryForwarderTest.java | 193 ++++++++++++++++++++-
.../replication/ReplicationLogTrackerTest.java | 43 +++++
5 files changed, 300 insertions(+), 5 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 d9b9d1e911..aac363add4 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
@@ -103,10 +103,45 @@ public class ReplicationLogDiscoveryForwarder extends
ReplicationLogDiscovery {
// 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();
+ FileSystem dstFS = remoteShardManager.getFileSystem();
Path dst = remoteShardManager.getWriterPath(ts, originServerName);
+ // Stage inside the shard's .staging subdirectory so replay never picks up
a half-written file
+ // and force-recovers the lease: every listing skips subdirectories via
FileStatus.isFile().
+ // Publish atomically via a same-shard rename up to dst once the bytes are
fully written.
+ Path staging = remoteShardManager.getStagingPath(dst);
long startTime = EnvironmentEdgeManager.currentTimeMillis();
- FileUtil.copy(srcFS, srcStat, remoteShardManager.getFileSystem(), dst,
false, false, conf);
- // successfully copied the file
+ // overwrite=true reclaims any orphan staging file left by a prior crashed
attempt of this same
+ // logical file (dst is keyed on the stable (ts, originServerName)). A
copy failure, or a failed
+ // rename when dst is absent, propagates: the source stays in out_progress
and is retried, and
+ // that retry re-copies with overwrite=true, reclaiming any partial
staging file left behind.
+ FileUtil.copy(srcFS, srcStat, dstFS, staging, false, true, conf);
+ if (!dstFS.rename(staging, dst)) {
+ // rename returned false. If dst already exists, a prior attempt of this
logical file
+ // published
+ // it and replay has not yet consumed it (the retry raced ahead of
replay). This is not
+ // exactly-once dedup: if replay already deleted dst, exists is false
and we throw so the
+ // source is retried, re-publishing dst (at-least-once; safe because
replay is idempotent).
+ if (dstFS.exists(dst)) {
+ LOG.info("Destination {} already present (retry raced ahead of replay)
for src={}", dst,
+ src);
+ // The publish is already complete, so cleaning up the redundant
staging file is best
+ // effort: this path returns normally and the source is then marked
completed, so there is
+ // no later retry to reclaim it via overwrite=true. A failure here
only leaks a staging
+ // file (invisible to replay); it must never demote an
already-delivered publish.
+ try {
+ if (!dstFS.delete(staging, false)) {
+ LOG.warn("Could not delete redundant staging file {} (dst {}
already published for "
+ + "src={}); it is orphaned and will not be reclaimed", staging,
dst, src);
+ }
+ } catch (IOException e) {
+ LOG.warn("Best-effort cleanup of staging file {} failed after dst {}
was published "
+ + "for src={}", staging, dst, src, e);
+ }
+ } else {
+ throw new IOException("Failed to rename staging file " + staging + "
to " + dst);
+ }
+ }
+ // successfully copied and published the file
long endTime = EnvironmentEdgeManager.currentTimeMillis();
long copyTime = endTime - startTime;
LOG.info("Copying file src={} dst={} size={} took {}ms", src, dst,
srcStat.getLen(), copyTime);
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 53395bb6d9..00bfa4ed68 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
@@ -415,8 +415,7 @@ public class ReplicationLogTracker {
* @return true if file format is correct, else false
*/
protected boolean isValidLogFile(Path file) {
- final String fileName = file.getName();
- return
fileName.endsWith(ReplicationShardDirectoryManager.LOG_FILE_EXTENSION);
+ return
file.getName().endsWith(ReplicationShardDirectoryManager.LOG_FILE_EXTENSION);
}
/**
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 174edc7ba9..bb77b52c65 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
@@ -67,6 +67,16 @@ public class ReplicationShardDirectoryManager {
/** File extension for replication log files. */
public static final String LOG_FILE_EXTENSION = ".plog";
+ /**
+ * Name of the staging subdirectory created inside each shard directory. The
forwarder copies an
+ * in-flight forwarded log file into
<shard>/.staging/<ts>_<server>.plog and
+ * then atomically renames it up to
<shard>/<ts>_<server>.plog once fully
+ * written. Because every replay/tracker listing gates on
+ * {@link org.apache.hadoop.fs.FileStatus#isFile()}, this subdirectory (and
any mid-copy file
+ * within it) is invisible to replay.
+ */
+ public static final String STAGING_SUB_DIRECTORY_NAME = ".staging";
+
/**
* Format string for log file names. <timestamp>_<servername>.plog Example
* 1762470665995_localhost,54575,1762470584502.plog
@@ -179,6 +189,23 @@ public class ReplicationShardDirectoryManager {
return new Path(shardPath, String.format(FILE_NAME_FORMAT, timestamp,
serverName));
}
+ /**
+ * Returns the staging path for a fully-resolved final writer path: {@code
+ * <shard>/.staging/<file-name>}. A file staged here keeps its real {@code
.plog} name but is
+ * invisible to replay because it lives under a subdirectory that every
replay/tracker listing
+ * skips via {@code FileStatus.isFile()}. The forwarder copies bytes here
and then atomically
+ * renames up to {@code finalPath} (same shard directory, same FileSystem)
to publish. The staging
+ * directory is created implicitly by the copy ({@code create()} makes
parent dirs), so this is a
+ * pure path computation with no namenode round trip.
+ * @param finalPath the final replay-eligible path (as returned by
+ * {@link #getWriterPath(long, String)})
+ * @return the staging path {@code <shard>/.staging/<file-name>}
+ */
+ public Path getStagingPath(Path finalPath) {
+ Path stagingDir = new Path(finalPath.getParent(),
STAGING_SUB_DIRECTORY_NAME);
+ return new Path(stagingDir, finalPath.getName());
+ }
+
/**
* Returns a ReplicationRound object based on the given round start time,
calculating the end time
* as start time + round duration.
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 e646155a9f..269e67605b 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
@@ -20,13 +20,18 @@ package org.apache.phoenix.replication;
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.STORE_AND_FORWARD;
import static
org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode.SYNC;
import static
org.apache.phoenix.replication.ReplicationShardDirectoryManager.PHOENIX_REPLICATION_ROUND_DURATION_SECONDS_KEY;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.HashSet;
import java.util.Optional;
@@ -39,6 +44,8 @@ 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.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -46,6 +53,7 @@ import org.apache.hadoop.hbase.client.Mutation;
import org.apache.phoenix.jdbc.HAGroupStoreRecord.HAGroupState;
import org.apache.phoenix.replication.ReplicationLogGroup.ReplicationMode;
import org.apache.phoenix.replication.log.LogFileTestUtil;
+import
org.apache.phoenix.replication.metrics.MetricsReplicationLogForwarderSourceFactory;
import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.junit.Assert;
import org.junit.Before;
@@ -188,7 +196,9 @@ public class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
FileSystem peerFs = peerShardManager.getFileSystem();
Set<String> peerNames = new HashSet<>();
for (FileStatus s : peerFs.listStatus(peerShardDir)) {
- peerNames.add(s.getPath().getName());
+ if (s.isFile()) {
+ peerNames.add(s.getPath().getName());
+ }
}
assertEquals("Both origin files should be forwarded to distinct
destinations", 2,
peerNames.size());
@@ -206,6 +216,187 @@ public class ReplicationLogDiscoveryForwarderTest extends
ReplicationLogBaseTest
return inProgress.get();
}
+ /** Reads the full contents of a file into a byte array. */
+ private byte[] readAll(FileSystem fs, Path file) throws IOException {
+ try (FSDataInputStream in = fs.open(file)) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ byte[] buf = new byte[4096];
+ int n;
+ while ((n = in.read(buf)) != -1) {
+ out.write(buf, 0, n);
+ }
+ return out.toByteArray();
+ }
+ }
+
+ /**
+ * Sets up a single in-progress source file on the local cluster and returns
its in-progress path.
+ * The peer shard manager is created eagerly so its FileSystem can be spied
by the caller.
+ */
+ private Path setUpSource(byte[] contents) throws IOException {
+ ReplicationLogTracker localTracker =
logGroup.getLogForwarder().getReplicationLogTracker();
+ ReplicationShardDirectoryManager localShardManager =
logGroup.getLocalShardManager();
+ long ts = EnvironmentEdgeManager.currentTimeMillis();
+ String origin = "10.244.1.10,16020,1784436416001";
+ Path shardDir = localShardManager.getShardDirectory(ts);
+ localFs.mkdirs(shardDir);
+ Path src = new Path(shardDir, ts + "_" + origin + ".plog");
+ try (FSDataOutputStream out = localFs.create(src, true)) {
+ out.write(contents);
+ }
+ Optional<Path> inProgress = localTracker.markInProgress(src);
+ assertTrue("markInProgress should succeed", inProgress.isPresent());
+ return inProgress.get();
+ }
+
+ /** The final .plog dst on the peer for the given local in-progress source
file. */
+ private Path peerDstFor(Path inProgressSrc) throws IOException {
+ ReplicationLogTracker localTracker =
logGroup.getLogForwarder().getReplicationLogTracker();
+ long ts = localTracker.getFileTimestamp(inProgressSrc);
+ String origin = localTracker.getServerName(inProgressSrc);
+ return logGroup.getOrCreatePeerShardManager().getWriterPath(ts, origin);
+ }
+
+ private Path stagingFor(Path dst) throws IOException {
+ return logGroup.getOrCreatePeerShardManager().getStagingPath(dst);
+ }
+
+ /**
+ * Installs a spy peer FileSystem so tests can intercept rename()/exists().
The peer shard manager
+ * (a spy) is returned by the spied logGroup, and its getFileSystem()
returns the spy FS. Callers
+ * use it to stub rename/exists and to assert on peer-side state.
+ */
+ private FileSystem installPeerFsSpy() throws IOException {
+ ReplicationShardDirectoryManager spyPeer =
spy(logGroup.getOrCreatePeerShardManager());
+ FileSystem peerFs = spy(spyPeer.getFileSystem());
+ doReturn(peerFs).when(spyPeer).getFileSystem();
+ doReturn(spyPeer).when(logGroup).getOrCreatePeerShardManager();
+ return peerFs;
+ }
+
+ /**
+ * Core regression guard for the cross-cluster lease race: while the
forwarder is copying bytes,
+ * the file must exist only under the non-replay-eligible .staging
subdirectory; the replay-
+ * eligible .plog appears in the shard directory only after the atomic
rename. We intercept
+ * rename() and assert the invariant at the moment just before the final
publish.
+ */
+ @Test
+ public void testForwardPublishesOnlyAfterRename() throws Exception {
+ byte[] contents = "some-log-bytes".getBytes();
+ Path src = setUpSource(contents);
+ Path dst = peerDstFor(src);
+ Path staging = stagingFor(dst);
+
+ ReplicationShardDirectoryManager peerShardManager =
logGroup.getOrCreatePeerShardManager();
+ FileSystem peerFs = installPeerFsSpy();
+ ReplicationLogTracker peerTracker =
+ new ReplicationLogTracker(conf, haGroupName, peerShardManager,
+
MetricsReplicationLogForwarderSourceFactory.getInstanceForTracker(haGroupName));
+
+ final boolean[] invariantHeld = { false };
+ doAnswer(new Answer<Boolean>() {
+ @Override
+ public Boolean answer(InvocationOnMock invocation) throws Throwable {
+ // At this point the bytes are fully staged but not yet published.
+ assertTrue("staging file must exist before rename",
peerFs.exists(staging));
+ assertFalse(".plog must NOT exist before rename", peerFs.exists(dst));
+ assertTrue("replay must see no eligible file while staging",
+ peerTracker.getNewFiles().isEmpty());
+ invariantHeld[0] = true;
+ return (Boolean) invocation.callRealMethod();
+ }
+ }).when(peerFs).rename(eq(staging), eq(dst));
+
+ logGroup.getLogForwarder().processFile(src);
+
+ assertTrue("rename interceptor should have run", invariantHeld[0]);
+ assertTrue(".plog must exist after rename", peerFs.exists(dst));
+ assertFalse("staging file must be gone after rename",
peerFs.exists(staging));
+ assertArrayEquals("published content must match source", contents,
readAll(peerFs, dst));
+ }
+
+ /**
+ * Retry after a crash between the rename and markCompleted (failure-mode
row #5): the final .plog
+ * already exists and replay has not yet consumed it. processFile must treat
this as
+ * already-delivered -- no throw, redundant staging dropped, dst untouched.
+ */
+ @Test
+ public void testForwardRetryOntoExistingDestinationSucceeds() throws
Exception {
+ byte[] contents = "retry-onto-existing".getBytes();
+ Path src = setUpSource(contents);
+ Path dst = peerDstFor(src);
+ Path staging = stagingFor(dst);
+
+ // The local test FileSystem's rename overwrites dst and returns true
(POSIX semantics), unlike
+ // HDFS which returns false when dst exists. Stub rename to return false
so we exercise the
+ // production already-delivered branch.
+ FileSystem peerFs = installPeerFsSpy();
+ doReturn(false).when(peerFs).rename(eq(staging), eq(dst));
+
+ // Simulate a prior successful publish of this same logical file.
+ byte[] existing = "already-there".getBytes();
+ try (FSDataOutputStream out = peerFs.create(dst, true)) {
+ out.write(existing);
+ }
+
+ logGroup.getLogForwarder().processFile(src);
+
+ assertTrue("dst should still exist", peerFs.exists(dst));
+ assertFalse("staging file should be cleaned up", peerFs.exists(staging));
+ assertArrayEquals("dst must be left untouched (not overwritten)", existing,
+ readAll(peerFs, dst));
+ }
+
+ /**
+ * Orphan reclamation (failure-mode rows #2/#3): a stale staging file left
by a crashed prior
+ * attempt is overwritten by the retry's copy (overwrite=true) and then
published.
+ */
+ @Test
+ public void testForwardReclaimsOrphanStagingFile() throws Exception {
+ byte[] contents = "fresh-content".getBytes();
+ Path src = setUpSource(contents);
+ Path dst = peerDstFor(src);
+ Path staging = stagingFor(dst);
+
+ FileSystem peerFs = logGroup.getOrCreatePeerShardManager().getFileSystem();
+ // Pre-create a stale/garbage staging file.
+ try (FSDataOutputStream out = peerFs.create(staging, true)) {
+ out.write("stale-garbage-bytes".getBytes());
+ }
+
+ logGroup.getLogForwarder().processFile(src);
+
+ assertTrue("dst should be published", peerFs.exists(dst));
+ assertFalse("staging file should be gone", peerFs.exists(staging));
+ assertArrayEquals("dst content must be the fresh source content, not the
orphan", contents,
+ readAll(peerFs, dst));
+ }
+
+ /**
+ * A genuine rename failure (rename returns false and dst does not exist)
must throw and leave dst
+ * uncreated so the source is retried from out_progress. The partial staging
file is left behind
+ * and reclaimed by the retry's overwrite=true copy.
+ */
+ @Test
+ public void testForwardRenameFailureLeavesSourceForRetry() throws Exception {
+ byte[] contents = "will-fail-rename".getBytes();
+ Path src = setUpSource(contents);
+ Path dst = peerDstFor(src);
+ Path staging = stagingFor(dst);
+
+ FileSystem peerFs = installPeerFsSpy();
+ doReturn(false).when(peerFs).rename(eq(staging), eq(dst));
+
+ try {
+ logGroup.getLogForwarder().processFile(src);
+ fail("processFile should have thrown on rename failure");
+ } catch (IOException expected) {
+ // expected
+ }
+
+ assertFalse("dst must not be created on failure", peerFs.exists(dst));
+ }
+
@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 2204f05b88..1f2fff3158 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
@@ -361,6 +361,49 @@ public class ReplicationLogTrackerTest {
assertEquals("File paths do not match", expectedPaths, actualPaths);
}
+ @Test
+ public void testStagingFilesExcludedFromListings() throws IOException {
+ // A file staged under the shard's .staging subdirectory must be invisible
to the shard-dir
+ // listings (getNewFilesForRound, getNewFiles). We assert two exclusions:
+ // 1. The .staging subdirectory (and the .plog file within it) is skipped
-- non-recursive
+ // listStatus never descends into it.
+ // 2. A subdirectory whose name is itself a valid <ts>_<server>.plog is
skipped by the
+ // FileStatus.isFile() gate. This is the case the gate uniquely catches:
without it, the
+ // directory entry would pass isValidLogFile() and the timestamp parse and
be listed.
+ tracker.init();
+
+ long roundStartTime = 1704153600000L;
+ long roundEndTime = roundStartTime + TimeUnit.MINUTES.toMillis(1);
+ ReplicationRound targetRound = new ReplicationRound(roundStartTime,
roundEndTime);
+ ReplicationShardDirectoryManager shardManager =
tracker.getReplicationShardDirectoryManager();
+ Path shardDirectory = shardManager.getShardDirectory(targetRound);
+ localFs.mkdirs(shardDirectory);
+
+ Path shardPublished = new Path(shardDirectory, "1704153600000_rs1.plog");
+ Path shardStaging =
+ shardManager.getStagingPath(new Path(shardDirectory,
"1704153630000_rs2.plog"));
+ // A directory named exactly like a valid published log file -- only
isFile() excludes it.
+ Path plogNamedDir = new Path(shardDirectory, "1704153640000_rs3.plog");
+ localFs.create(shardPublished, true).close();
+ localFs.create(shardStaging, true).close();
+ localFs.mkdirs(plogNamedDir);
+
+ assertListedNames("getNewFilesForRound",
tracker.getNewFilesForRound(targetRound),
+ shardPublished, shardStaging, plogNamedDir);
+ assertListedNames("getNewFiles", tracker.getNewFiles(), shardPublished,
shardStaging,
+ plogNamedDir);
+ }
+
+ /** Asserts a listing contains the published file's name but none of the
excluded names. */
+ private void assertListedNames(String method, List<Path> result, Path
published,
+ Path... excluded) {
+ Set<String> names =
result.stream().map(Path::getName).collect(Collectors.toSet());
+ assertTrue(method + " must list the published .plog",
names.contains(published.getName()));
+ for (Path ex : excluded) {
+ assertFalse(method + " must not list " + ex.getName(),
names.contains(ex.getName()));
+ }
+ }
+
@Test
public void testGetInProgressFiles() throws IOException {
// Initialize tracker
