PHOENIX-7977 Stage forwarded replication log files in a per-shard staging subdirectory before atomic rename

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);

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);
   }
 
   /**

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 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
+   * <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.

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,13 +44,16 @@
 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;
 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 void testForwardPreservesOriginServerIdentity() throws Exception {
     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 @@ 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<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 };

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 <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