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 d9b9d1e9119..aac363add47 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 @@ protected void processFile(Path src) throws IOException { // 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 53395bb6d9b..00bfa4ed685 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 @@ private String buildInProgressFileName(String prefix) { * @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 174edc7ba96..bb77b52c659 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. _.plog Example * 1762470665995_localhost,54575,1762470584502.plog @@ -179,6 +189,23 @@ public Path getWriterPath(long timestamp, String serverName) throws IOException return new Path(shardPath, String.format(FILE_NAME_FORMAT, timestamp, serverName)); } + /** + * Returns the staging path for a fully-resolved final writer path: {@code + * /.staging/}. 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 /.staging/} + */ + 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 e646155a9ff..269e67605b8 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 @@ 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.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.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 void testForwardPreservesOriginServerIdentity() throws Exception { FileSystem peerFs = peerShardManager.getFileSystem(); Set 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 @@ private Path markInProgressSource(ReplicationLogTracker tracker, Path file) thro 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 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() { + @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 2204f05b887..1f2fff31585 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 void testGetNewFilesForRoundWithInvalidFiles() throws IOException { 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 _.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 result, Path published, + Path... excluded) { + Set 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