HDDS-16460. Clear recovering entries when removing containers

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerSet.java

@@ -368,7 +368,7 @@ public boolean removeMissingContainer(long containerId) throws StorageContainerE
   }
 
   /**
-   * Removes the Container matching with specified containerId.
+   * Removes the container matching the specified containerId and its recovering entry, if present.
    * @param containerId ID of the container to remove
    * @return If container is removed from containerMap returns true, otherwise
    * false
@@ -383,6 +383,7 @@ private boolean removeContainer(long containerId, boolean markMissing, boolean r
     if (markMissing) {
       missingContainerSet.add(containerId);
     }
+    removeRecoveringContainer(containerId);
     Container<?> removed = containerMap.remove(containerId);
     if (removeFromDB) {
       deleteFromContainerTable(containerId);

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerSet.java

@@ -143,6 +143,31 @@ public void testAddGetRemoveContainer(ContainerLayoutVersion layout)
     assertFalse(containerSet.removeContainer(1000L));
   }
 
+  @ContainerLayoutTestInfo.ContainerTest
+  public void testRemoveRecoveringContainerFromMemory(ContainerLayoutVersion layout) throws StorageContainerException {
+    ContainerSet containerSet = newContainerSet();
+    long containerId = 100L;
+    KeyValueContainerData data = new KeyValueContainerData(containerId, layout,
+        (long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(), UUID.randomUUID().toString());
+    data.setState(ContainerProtos.ContainerDataProto.State.RECOVERING);
+    KeyValueContainer container = new KeyValueContainer(data, new OzoneConfiguration());
+    containerSet.addContainer(container);
+    assertThat(containerSet.getRecoveringContainerMap()).containsKey(containerId);
+
+    assertThat(containerSet.removeContainerOnlyFromMemory(containerId)).isTrue();
+    assertThat(containerSet.getContainer(containerId)).isNull();
+    assertThat(containerSet.getRecoveringContainerMap()).doesNotContainKey(containerId);
+
+    containerSet.addContainer(container);
+    assertThat(containerSet.getRecoveringContainerMap()).containsKey(containerId);
+    containerSet.removeContainerOnlyFromMemory(containerId);
+    data.closeContainer();
+    containerSet.addContainer(container);
+    assertThat(containerSet.getContainer(containerId).getContainerState())
+        .isEqualTo(ContainerProtos.ContainerDataProto.State.CLOSED);
+    assertThat(containerSet.getRecoveringContainerMap()).doesNotContainKey(containerId);
+  }
+
   @ContainerLayoutTestInfo.ContainerTest
   public void testIteratorsAndCount(ContainerLayoutVersion layout)
       throws StorageContainerException {

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java

@@ -44,6 +44,7 @@
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.net.HostAndPort;
@@ -65,6 +66,8 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 
 /**
  * Verify that {@link MutableVolumeSet} correctly checks for failed disks
@@ -248,8 +251,9 @@ private OzoneConfiguration getConfWithDataNodeDirs(int numDirs) {
    * And FCR report is being sent.
    * @throws IOException
    */
-  @Test
-  public void testVolumeFailure() throws IOException {
+  @ParameterizedTest
+  @EnumSource(value = State.class, names = {"CLOSED", "RECOVERING"})
+  public void testVolumeFailure(State state) throws IOException {
     final int numVolumes = 5;
 
     conf = getConfWithDataNodeDirs(numVolumes);
@@ -276,15 +280,15 @@ public void testVolumeFailure() throws IOException {
         new KeyValueContainerData(containerID, layout,
             ContainerTestHelper.CONTAINER_MAX_SIZE,
             UUID.randomUUID().toString(), datanodeId.toString());
-    data.closeContainer();
+    data.setState(state);
     data.setSchemaVersion(OzoneConsts.SCHEMA_V3);
 
     long containerID1 = ContainerTestHelper.getTestContainerID();
     KeyValueContainerData data1 =
         new KeyValueContainerData(containerID1, layout,
             ContainerTestHelper.CONTAINER_MAX_SIZE,
             UUID.randomUUID().toString(), datanodeId.toString());
-    data1.closeContainer();
+    data1.setState(state);
     data1.setSchemaVersion(OzoneConsts.SCHEMA_V3);
 
     final MutableVolumeSet volumeSet = new MutableVolumeSet(
@@ -310,6 +314,7 @@ public void testVolumeFailure() throws IOException {
     container1.create(volumeSet1,
         new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString());
     conSet.addContainer(container1);
+    assertThat(conSet.getRecoveringContainerMap()).hasSize(state == State.RECOVERING ? 2 : 0);
     DatanodeStateMachine datanodeStateMachineMock =
         mock(DatanodeStateMachine.class);
     StateContext stateContext = new StateContext(
@@ -332,8 +337,10 @@ public void testVolumeFailure() throws IOException {
     // ContainerID1 should be removed belonging to failed volume
     assertNull(conSet.getContainer(containerID1));
     assertTrue(conSet.getMissingContainerSet().contains(containerID1));
+    assertThat(conSet.getRecoveringContainerMap()).doesNotContainKey(containerID1);
     // ContainerID should exist belonging to normal volume
     assertNotNull(conSet.getContainer(containerID));
+    assertThat(conSet.getRecoveringContainerMap()).hasSize(state == State.RECOVERING ? 1 : 0);
     expectedReportCount.put(
         StorageContainerDatanodeProtocolProtos.ContainerReportsProto
             .getDescriptor().getFullName(), 1);

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

@@ -627,6 +627,32 @@ public void testDeleteContainer() throws IOException {
     }
   }
 
+  @Test
+  public void testDeleteRecoveringContainer() throws IOException {
+    HandlerWithVolumeSet handlerCtx = createKeyValueHandler(tempDir);
+    KeyValueHandler keyValueHandler = handlerCtx.getHandler();
+    ContainerSet containerSet = handlerCtx.getContainerSet();
+    try {
+      ContainerCommandRequestProto.Builder request = createContainerRequest(DATANODE_UUID, DUMMY_CONTAINER_ID)
+          .toBuilder();
+      request.getCreateContainerBuilder().setState(State.RECOVERING);
+      assertThat(keyValueHandler.handleCreateContainer(request.build(), null).getResult())
+          .isEqualTo(ContainerProtos.Result.SUCCESS);
+      Container<?> container = containerSet.getContainer(DUMMY_CONTAINER_ID);
+      assertThat(container).isNotNull();
+      assertThat(container.getContainerState()).isEqualTo(State.RECOVERING);
+      assertThat(containerSet.getRecoveringContainerMap()).containsKey(DUMMY_CONTAINER_ID);
+
+      keyValueHandler.deleteContainer(container, true);
+
+      assertThat(containerSet.getContainer(DUMMY_CONTAINER_ID)).isNull();
+      assertThat(containerSet.getRecoveringContainerMap()).doesNotContainKey(DUMMY_CONTAINER_ID);
+    } finally {
+      keyValueHandler.stop();
+      handlerCtx.getVolumeSet().getVolumesList().forEach(StorageVolume::shutdown);
+    }
+  }
+
   /**
    * Tests that deleting a container decrements the cached used space of its volume.
    */