HDDS-13070. OM Follower changes to create and place sst files from hardlink file.

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMDbCheckpointServletInodeBasedXfer.java

@@ -227,7 +227,6 @@ void testContentsOfTarballWithSnapshot() throws Exception {
         .forEachRemaining(snapInfo -> snapshotPaths.add(getSnapshotDBPath(snapInfo.getCheckpointDir())));
     Set<String> inodesFromOmDataDir = new HashSet<>();
     Set<String> inodesFromTarball = new HashSet<>();
-    Set<Path> allPathsInTarball = new HashSet<>();
     try (Stream<Path> filesInTarball = Files.list(newDbDir.toPath())) {
       List<Path> files = filesInTarball.collect(Collectors.toList());
       for (Path p : files) {
@@ -237,7 +236,6 @@ void testContentsOfTarballWithSnapshot() throws Exception {
         }
         String inode = getInode(file.getName());
         inodesFromTarball.add(inode);
-        allPathsInTarball.add(p);
       }
     }
     Map<String, List<String>> hardLinkMapFromOmData = new HashMap<>();
@@ -269,10 +267,7 @@ void testContentsOfTarballWithSnapshot() throws Exception {
     assertTrue(inodesFromTarball.containsAll(inodesFromOmDataDir));
 
     // create hardlinks now
-    OmSnapshotUtils.createHardLinks(newDbDir.toPath());
-    for (Path old : allPathsInTarball) {
-      assertTrue(old.toFile().delete());
-    }
+    OmSnapshotUtils.createHardLinks(newDbDir.toPath(), true);
     assertFalse(hardlinkFilePath.toFile().exists());
   }
 
@@ -301,7 +296,7 @@ public void testSnapshotDBConsistency() throws Exception {
     FileUtil.unTar(tempFile, newDbDir);
     Set<Path> allPathsInTarball = getAllPathsInTarball(newDbDir);
     // create hardlinks now
-    OmSnapshotUtils.createHardLinks(newDbDir.toPath());
+    OmSnapshotUtils.createHardLinks(newDbDir.toPath(), false);
     for (Path old : allPathsInTarball) {
       assertTrue(old.toFile().delete());
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -3980,7 +3980,7 @@ public synchronized TermIndex installSnapshotFromLeader(String leaderId) {
     TermIndex termIndex = null;
     try {
       // Install hard links.
-      OmSnapshotUtils.createHardLinks(omDBCheckpoint.getCheckpointLocation());
+      OmSnapshotUtils.createHardLinks(omDBCheckpoint.getCheckpointLocation(), false);
       termIndex = installCheckpoint(leaderId, omDBCheckpoint);
     } catch (Exception ex) {
       LOG.error("Failed to install snapshot from Leader OM.", ex);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManagerHttpServer.java

@@ -36,7 +36,7 @@ public OzoneManagerHttpServer(MutableConfigurationSource conf,
     addServlet("serviceList", OZONE_OM_SERVICE_LIST_HTTP_ENDPOINT,
         ServiceListJSONServlet.class);
     addServlet("dbCheckpoint", OZONE_DB_CHECKPOINT_HTTP_ENDPOINT,
-        OMDBCheckpointServlet.class);
+        OMDBCheckpointServletInodeBasedXfer.class);
     getWebAppContext().setAttribute(OzoneConsts.OM_CONTEXT_ATTRIBUTE, om);
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OmSnapshotUtils.java

@@ -28,11 +28,14 @@
 import java.nio.file.Paths;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.nio.file.attribute.FileTime;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.hadoop.ozone.om.OmSnapshotManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Ozone Manager Snapshot Utilities.
@@ -41,6 +44,8 @@ public final class OmSnapshotUtils {
 
   public static final String DATA_PREFIX = "data";
   public static final String DATA_SUFFIX = "txt";
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OmSnapshotUtils.class);
 
   private OmSnapshotUtils() { }
 
@@ -124,20 +129,28 @@ public static Path createHardLinkList(int truncateLength,
    * Create hard links listed in OM_HARDLINK_FILE.
    *
    * @param dbPath Path to db to have links created.
+   * @param deleteSourceFiles - Whether to delete the source files after creating the links.
    */
-  public static void createHardLinks(Path dbPath) throws IOException {
+  public static void createHardLinks(Path dbPath, boolean deleteSourceFiles) throws IOException {
     File hardLinkFile =
         new File(dbPath.toString(), OmSnapshotManager.OM_HARDLINK_FILE);
+    List<Path> filesToDelete = new ArrayList<>();
     if (hardLinkFile.exists()) {
       // Read file.
       try (Stream<String> s = Files.lines(hardLinkFile.toPath())) {
         List<String> lines = s.collect(Collectors.toList());
 
         // Create a link for each line.
         for (String l : lines) {
-          String from = l.split("\t")[1];
-          String to = l.split("\t")[0];
+          String[] parts = l.split("\t");
+          if (parts.length != 2) {
+            LOG.warn("Skipping malformed line in hardlink file: {}", l);
+            return;
+          }
+          String from = parts[1];
+          String to = parts[0];
           Path fullFromPath = Paths.get(dbPath.toString(), from);
+          filesToDelete.add(fullFromPath);
           Path fullToPath = Paths.get(dbPath.toString(), to);
           // Make parent dir if it doesn't exist.
           Path parent = fullToPath.getParent();
@@ -154,6 +167,15 @@ public static void createHardLinks(Path dbPath) throws IOException {
         }
       }
     }
+    if (deleteSourceFiles) {
+      for (Path fileToDelete : filesToDelete) {
+        try {
+          Files.deleteIfExists(fileToDelete);
+        } catch (IOException e) {
+          LOG.warn("Couldn't delete source file {} while unpacking the DB", fileToDelete, e);
+        }
+      }
+    }
   }
 
   /**

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshotManager.java

@@ -374,7 +374,7 @@ public void testHardLinkCreation() throws IOException {
     File s1FileLink = new File(followerSnapDir2, "s1.sst");
 
     // Create links on the follower from list.
-    OmSnapshotUtils.createHardLinks(candidateDir.toPath());
+    OmSnapshotUtils.createHardLinks(candidateDir.toPath(), false);
 
     // Confirm expected follower links.
     assertTrue(s1FileLink.exists());