HDDS-16880. Recon should not retain incomplete OM DB checkpoint

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java

@@ -193,7 +193,7 @@ public DBCheckpoint getCheckpointFromUntarredDb(Path untarredDbDir) throws IOExc
    * @return  path of untarred dbDir.
    * @throws IOException
    */
-  private Path untarContentsOfTarball(File snapshot,
+  protected Path untarContentsOfTarball(File snapshot,
       File untarDir, boolean deleteSnapshot) throws IOException {
     // Untar the checkpoint file.
     Path untarredDbDir = untarDir.toPath();

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java

@@ -508,8 +508,8 @@ public DBCheckpoint getOzoneManagerDBSnapshot() {
       // Do not wipe the candidate dir here. The shared RDBSnapshotProvider's
       // checkLeaderConsistency() manages it on the next attempt: kept for the
       // same leader (so a future batched/chunked transfer can resume the parts
-      // already received), reset on a leader change. This matches how the OM
-      // follower handles a failed download.
+      // already received), reset on a leader change. It only holds fully
+      // extracted files, see ReconRDBSnapshotProvider#untarContentsOfTarball.
       return null;
     }
   }

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconRDBSnapshotProvider.java

@@ -37,11 +37,15 @@
 import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
+import org.apache.commons.compress.archivers.ArchiveInputStream;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hdds.server.http.HttpConfig;
+import org.apache.hadoop.hdds.utils.Archiver;
 import org.apache.hadoop.hdds.utils.HAUtils;
 import org.apache.hadoop.hdds.utils.RDBSnapshotProvider;
 import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
@@ -164,6 +168,36 @@ public void downloadSnapshot(String leaderNodeID, File targetFile)
     });
   }
 
+  /**
+   * Extract each file to a temporary name and move it to its own name only once all its bytes are written.
+   * The candidate dir is kept after a failed fetch and the next request lists its files as already received,
+   * so an untar that fails part way (a truncated stream or a full disk) must not leave a partly written file there.
+   */
+  @Override
+  protected Path untarContentsOfTarball(File snapshot, File untarDir, boolean deleteSnapshot) throws IOException {
+    Path dir = untarDir.toPath();
+    try (ArchiveInputStream<TarArchiveEntry> in = Archiver.untar(Files.newInputStream(snapshot.toPath()))) {
+      for (TarArchiveEntry entry = in.getNextEntry(); entry != null; entry = in.getNextEntry()) {
+        Path path = dir.resolve(entry.getName());
+        if (entry.isDirectory()) {
+          Archiver.extractEntry(entry, in, entry.getSize(), dir, path);
+        } else {
+          Path tmp = path.resolveSibling(path.getFileName() + ".tmp");
+          try {
+            Archiver.extractEntry(entry, in, entry.getSize(), dir, tmp);
+            Files.move(tmp, path, StandardCopyOption.REPLACE_EXISTING);
+          } finally {
+            Files.deleteIfExists(tmp);
+          }
+        }
+      }
+    }
+    if (deleteSnapshot) {
+      FileUtils.deleteQuietly(snapshot);
+    }
+    return dir;
+  }
+
   /**
    * After the transfer completes, install the leader's hard-link inventory,
    * normalize the layout to {@code <candidate>/om.db}, then move that DB out of

hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java

@@ -18,6 +18,8 @@
 package org.apache.hadoop.ozone.recon.spi.impl;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
 import static org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.getTestReconOmMetadataManager;
 import static org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.initializeEmptyOmMetadataManager;
 import static org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.initializeNewOmMetadataManager;
@@ -29,6 +31,7 @@
 import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.RECON_OM_DELTA_UPDATE_LIMIT;
 import static org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl.OmSnapshotTaskName.OmDeltaRequest;
 import static org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl.OmSnapshotTaskName.OmSnapshotRequest;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -48,12 +51,20 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import org.apache.commons.compress.archivers.ArchiveOutputStream;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.server.http.HttpConfig;
+import org.apache.hadoop.hdds.utils.Archiver;
+import org.apache.hadoop.hdds.utils.HAUtils;
 import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
 import org.apache.hadoop.hdds.utils.db.DBStore;
 import org.apache.hadoop.hdds.utils.db.RDBStore;
@@ -327,6 +338,47 @@ public void testGetOzoneManagerDBSnapshot(@TempDir File dirReconMetadata)
     assertEquals(2, files.length);
   }
 
+  @Test
+  public void testPartiallyUntarredFileNotReportedOnRetry(@TempDir File dirReconMetadata) throws Exception {
+    File completeSst = new File(dirReconMetadata, "000009.sst");
+    File truncatedSst = new File(dirReconMetadata, "000010.sst");
+    FileUtils.writeByteArrayToFile(completeSst, new byte[10_000]);
+    FileUtils.writeByteArrayToFile(truncatedSst, new byte[100_000]);
+    List<List<String>> requestedExclusions = new ArrayList<>();
+    ReconRDBSnapshotProvider snapshotProvider = new ReconRDBSnapshotProvider(omSnapshotDbParentDir, null, false,
+        HttpConfig.Policy.HTTP_ONLY, false, true, () -> null) {
+      @Override
+      public void downloadSnapshot(String leaderNodeID, File targetFile) throws IOException {
+        // The files the request tells the leader not to send again.
+        requestedExclusions.add(HAUtils.getExistingFiles(getCandidateDir()));
+        // The leader stream ends in the middle of the second, larger SST file.
+        try (ArchiveOutputStream<TarArchiveEntry> out = Archiver.tar(Files.newOutputStream(targetFile.toPath()))) {
+          Archiver.includeFile(completeSst, completeSst.getName(), out);
+          Archiver.includeFile(truncatedSst, truncatedSst.getName(), out);
+        }
+        try (RandomAccessFile file = new RandomAccessFile(targetFile, "rw")) {
+          file.setLength(file.length() / 2);
+        }
+      }
+    };
+    ReconUtils reconUtilsMock = getMockReconUtils();
+    when(reconUtilsMock.getReconNodeDetails(any(OzoneConfiguration.class)))
+        .thenReturn(ReconTestUtils.getReconNodeDetails());
+    OzoneManagerServiceProviderImpl ozoneManagerServiceProvider = new OzoneManagerServiceProviderImpl(configuration,
+        mock(ReconOMMetadataManager.class), getMockTaskController(), reconUtilsMock, ozoneManagerProtocol,
+        reconContext, getMockTaskStatusUpdaterManager());
+    ozoneManagerServiceProvider.setReconSnapshotProvider(snapshotProvider);
+
+    assertNull(ozoneManagerServiceProvider.getOzoneManagerDBSnapshot());
+    assertNull(ozoneManagerServiceProvider.getOzoneManagerDBSnapshot());
+
+    // The fully extracted file is kept and listed as received, so the leader does not send it again.
+    // The truncated file must not be listed, or the leader would skip it too.
+    assertThat(requestedExclusions).containsExactly(emptyList(), singletonList(completeSst.getName()));
+    assertThat(snapshotProvider.getCandidateDir().list()).containsExactly(completeSst.getName());
+    assertEquals(completeSst.length(), new File(snapshotProvider.getCandidateDir(), completeSst.getName()).length());
+  }
+
   static RocksDatabase getRocksDatabase(OMMetadataManager om) {
     return ((RDBStore)om.getStore()).getDb();
   }