HDDS-5032. DN stopped to load containers on volume a container load exception

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerReader.java

@@ -150,14 +150,20 @@ public boolean accept(File pathname) {
             File[] containerDirs = containerTopDir.listFiles();
             if (containerDirs != null) {
               for (File containerDir : containerDirs) {
-                File containerFile = ContainerUtils.getContainerFile(
-                    containerDir);
-                long containerID = ContainerUtils.getContainerID(containerDir);
-                if (containerFile.exists()) {
-                  verifyContainerFile(storageLoc, containerID, containerFile);
-                } else {
-                  LOG.error("Missing .container file for ContainerID: {}",
-                      containerDir.getName());
+                try {
+                  File containerFile = ContainerUtils.getContainerFile(
+                      containerDir);
+                  long containerID =
+                      ContainerUtils.getContainerID(containerDir);
+                  if (containerFile.exists()) {
+                    verifyContainerFile(storageLoc, containerID, containerFile);
+                  } else {
+                    LOG.error("Missing .container file for ContainerID: {}",
+                        containerDir.getName());
+                  }
+                } catch (Throwable e) {
+                  LOG.error("Failed to load container from {}",
+                      containerDir.getAbsolutePath(), e);
                 }
               }
             }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java

@@ -213,6 +213,49 @@ public void testContainerReader() throws Exception {
     }
   }
 
+  @Test
+  public void testContainerReaderWithLoadException() throws Exception {
+    MutableVolumeSet volumeSet1;
+    HddsVolume hddsVolume1;
+    ContainerSet containerSet1 = new ContainerSet();
+    File volumeDir1 = tempDir.newFolder();
+    RoundRobinVolumeChoosingPolicy volumeChoosingPolicy1;
+
+    volumeSet1 = Mockito.mock(MutableVolumeSet.class);
+    UUID datanode = UUID.randomUUID();
+    hddsVolume1 = new HddsVolume.Builder(volumeDir1
+        .getAbsolutePath()).conf(conf).datanodeUuid(datanode
+        .toString()).clusterID(clusterId).build();
+    volumeChoosingPolicy1 = mock(RoundRobinVolumeChoosingPolicy.class);
+    Mockito.when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong()))
+        .thenReturn(hddsVolume1);
+
+    int containerCount = 3;
+    for (int i = 0; i < containerCount; i++) {
+      KeyValueContainerData keyValueContainerData = new KeyValueContainerData(i,
+          ChunkLayOutVersion.FILE_PER_BLOCK,
+          (long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(),
+          datanodeId.toString());
+      KeyValueContainer keyValueContainer =
+          new KeyValueContainer(keyValueContainerData, conf);
+      keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId);
+      BlockUtils.removeDB(keyValueContainerData, conf);
+
+      if (i == 0) {
+        // rename first container directory name
+        String containerPathStr =
+            keyValueContainer.getContainerData().getContainerPath();
+        File containerPath = new File(containerPathStr);
+        String renamePath = containerPathStr + "-aa";
+        containerPath.renameTo(new File(renamePath));
+      }
+    }
+
+    ContainerReader containerReader = new ContainerReader(volumeSet1,
+        hddsVolume1, containerSet1, conf);
+    containerReader.readVolume(hddsVolume1.getHddsRootDir());
+    Assert.assertEquals(containerCount - 1, containerSet1.containerCount());
+  }
   @Test
   public void testMultipleContainerReader() throws Exception {
     final int volumeNum = 10;