diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java index 8253f7bb1124..238fe6c706ea 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java +++ b/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(); diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java index 9cc4d0e15bf4..ed9a9e49233c 100644 --- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java +++ b/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; } } diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconRDBSnapshotProvider.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconRDBSnapshotProvider.java index 7241178e9b9c..2f5b7e1df616 100644 --- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconRDBSnapshotProvider.java +++ b/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 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 /om.db}, then move that DB out of diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java index 97150b703f4f..70aa7c104c33 100644 --- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java +++ b/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> 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 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(); }