HDDS-6235. Empty KeyValueContainers are replicated without chunks directory.

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java

@@ -510,6 +510,12 @@ public void importContainerData(InputStream input,
       //fill in memory stat counter (keycount, byte usage)
       KeyValueContainerUtil.parseKVContainerData(containerData, config);
 
+      if (!scanMetaData()) {
+        String message = "Metadata scan of imported container " +
+            containerData.getContainerID() + " failed. Attempting clean up.";
+        LOG.error(message);
+        throw new IOException(message);
+      }
     } catch (Exception ex) {
       if (ex instanceof StorageContainerException &&
           ((StorageContainerException) ex).getResult() ==
@@ -551,6 +557,15 @@ public void exportContainerData(OutputStream destination,
                 " is in state " + state);
       }
 
+      if (!scanMetaData()) {
+        markContainerUnhealthy();
+        String message = "Metadata scan failed before exporting " +
+            "container " + containerData.getContainerID() + ". Container " +
+            "export aborted and container marked unhealthy.";
+        LOG.error(message);
+        throw new IOException(message);
+      }
+
       try {
         compactDB();
         // Close DB (and remove from cache) to avoid concurrent modification

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/TarContainerPacker.java

@@ -84,11 +84,19 @@ public byte[] unpackContainerData(Container<KeyValueContainerData> container,
         if (name.startsWith(DB_DIR_NAME + "/")) {
           Path destinationPath = dbRoot
               .resolve(name.substring(DB_DIR_NAME.length() + 1));
-          extractEntry(archiveInput, size, dbRoot, destinationPath);
+          if (entry.isDirectory()) {
+            Files.createDirectories(destinationPath);
+          } else {
+            extractFileEntry(archiveInput, size, dbRoot, destinationPath);
+          }
         } else if (name.startsWith(CHUNKS_DIR_NAME + "/")) {
           Path destinationPath = chunksRoot
               .resolve(name.substring(CHUNKS_DIR_NAME.length() + 1));
-          extractEntry(archiveInput, size, chunksRoot, destinationPath);
+          if (entry.isDirectory()) {
+            Files.createDirectories(destinationPath);
+          } else {
+            extractFileEntry(archiveInput, size, chunksRoot, destinationPath);
+          }
         } else if (CONTAINER_FILE_NAME.equals(name)) {
           //Don't do anything. Container file should be unpacked in a
           //separated step by unpackContainerDescriptor call.
@@ -109,8 +117,8 @@ public byte[] unpackContainerData(Container<KeyValueContainerData> container,
     }
   }
 
-  private void extractEntry(InputStream input, long size,
-                            Path ancestor, Path path) throws IOException {
+  private void extractFileEntry(InputStream input, long size,
+      Path ancestor, Path path) throws IOException {
     HddsUtils.validatePath(path, ancestor);
     Path parent = path.getParent();
     if (parent != null) {
@@ -209,6 +217,12 @@ private byte[] readEntry(InputStream input, final long size)
   private void includePath(Path dir, String subdir,
       ArchiveOutputStream archiveOutput) throws IOException {
 
+    // Add a directory entry before adding files, in case the directory is
+    // empty.
+    ArchiveEntry entry = archiveOutput.createArchiveEntry(dir.toFile(), subdir);
+    archiveOutput.putArchiveEntry(entry);
+
+    // Add files in the directory.
     try (Stream<Path> dirEntries = Files.list(dir)) {
       for (Path path : dirEntries.collect(toList())) {
         String entryName = subdir + "/" + path.getFileName();

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerUtil.java

@@ -19,6 +19,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
@@ -254,6 +255,13 @@ public static void parseKVContainerData(KeyValueContainerData kvContainerData,
       if (!isBlockMetadataSet) {
         initializeUsedBytesAndBlockCount(store, kvContainerData);
       }
+
+      // If the container is missing a chunks directory, possibly due to the
+      // bug fixed by HDDS-6235, create it here.
+      File chunksDir = new File(kvContainerData.getChunksPath());
+      if (!chunksDir.exists()) {
+        Files.createDirectories(chunksDir.toPath());
+      }
     } finally {
       if (cachedDB != null) {
         // If we get a cached instance, calling close simply decrements the

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java

@@ -66,6 +66,8 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.List;
@@ -74,6 +76,7 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_DB_PROFILE;
 import static org.apache.ratis.util.Preconditions.assertTrue;
@@ -154,6 +157,36 @@ public void testCreateContainer() throws Exception {
         "DB does not exist");
   }
 
+
+  @Test
+  public void testEmptyContainerImportExport() throws Exception {
+    createContainer();
+    closeContainer();
+
+    KeyValueContainerData data = keyValueContainer.getContainerData();
+
+    // Check state of original container.
+    checkContainerFilesPresent(data, 0);
+
+    //destination path
+    File exportTar = folder.newFile("exported.tar.gz");
+    TarContainerPacker packer = new TarContainerPacker();
+    //export the container
+    try (FileOutputStream fos = new FileOutputStream(exportTar)) {
+      keyValueContainer.exportContainerData(fos, packer);
+    }
+
+    keyValueContainer.delete();
+
+    // import container.
+    try (FileInputStream fis = new FileInputStream(exportTar)) {
+      keyValueContainer.importContainerData(fis, packer);
+    }
+
+    // Make sure empty chunks dir was unpacked.
+    checkContainerFilesPresent(data, 0);
+  }
+
   @Test
   public void testContainerImportExport() throws Exception {
     long containerId = keyValueContainer.getContainerData().getContainerID();
@@ -244,6 +277,18 @@ public void testContainerImportExport() throws Exception {
     }
   }
 
+  private void checkContainerFilesPresent(KeyValueContainerData data,
+      long expectedNumFilesInChunksDir) throws IOException {
+    File chunksDir = new File(data.getChunksPath());
+    Assert.assertTrue(Files.isDirectory(chunksDir.toPath()));
+    try (Stream<Path> stream = Files.list(chunksDir.toPath())) {
+      Assert.assertEquals(expectedNumFilesInChunksDir, stream.count());
+    }
+    Assert.assertTrue(data.getDbFile().exists());
+    Assert.assertTrue(KeyValueContainer.getContainerFile(data.getMetadataPath(),
+        data.getContainerID()).exists());
+  }
+
   /**
    * Create the container on disk.
    */