HDDS-8838. Update default datanode check empty containter on disk to false

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java

@@ -86,13 +86,9 @@ public class DatanodeConfiguration {
   public static final String
       ROCKSDB_DELETE_OBSOLETE_FILES_PERIOD_MICRO_SECONDS_KEY =
       "hdds.datanode.rocksdb.delete_obsolete_files_period";
-
-  public static final String
-      OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE =
-      "hdds.datanode.check.empty.container.delete";
   public static final Boolean
-      OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE_DEFAULT =
-      true;
+      OZONE_DATANODE_CHECK_EMPTY_CONTAINER_DIR_ON_DELETE_DEFAULT =
+      false;
 
   /**
    * Number of threads per volume that Datanode will use for chunk read.
@@ -412,6 +408,20 @@ public void setWaitOnAllFollowers(boolean val) {
   )
   private int autoCompactionSmallSstFileNum = 512;
 
+  /**
+   * Whether to check container directory or not to determine
+   * container is empty.
+   */
+  @Config(key = "hdds.datanode.check.empty.container.dir.on.delete",
+      type = ConfigType.BOOLEAN,
+      defaultValue = "false",
+      tags = { DATANODE },
+      description = "Boolean Flag to decide whether to check container " +
+          "directory or not to determine container is empty"
+  )
+  private boolean bCheckEmptyContainerDir =
+      OZONE_DATANODE_CHECK_EMPTY_CONTAINER_DIR_ON_DELETE_DEFAULT;
+
   @PostConstruct
   public void validate() {
     if (containerDeleteThreads < 1) {
@@ -531,6 +541,10 @@ public void setFailedDbVolumesTolerated(int failedVolumesTolerated) {
     this.failedDbVolumesTolerated = failedVolumesTolerated;
   }
 
+  public boolean getCheckEmptyContainerDir() {
+    return bCheckEmptyContainerDir;
+  }
+
   public Duration getDiskCheckMinGap() {
     return Duration.ofMillis(diskCheckMinGap);
   }

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

@@ -33,6 +33,7 @@
 import java.util.Set;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdds.HddsUtils;
@@ -109,7 +110,7 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
   // container are synchronous.
   private Set<Long> pendingPutBlockCache;
 
-  private final boolean bCheckChunksFilePath;
+  private boolean bCheckChunksFilePath;
 
   public KeyValueContainer(KeyValueContainerData containerData,
       ConfigurationSource ozoneConfig) {
@@ -126,12 +127,14 @@ public KeyValueContainer(KeyValueContainerData containerData,
     } else {
       this.pendingPutBlockCache = Collections.emptySet();
     }
-    bCheckChunksFilePath =
-        ozoneConfig.getBoolean(
-            DatanodeConfiguration.
-                OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE,
-            DatanodeConfiguration.
-                OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE_DEFAULT);
+    DatanodeConfiguration dnConf =
+        config.getObject(DatanodeConfiguration.class);
+    bCheckChunksFilePath = dnConf.getCheckEmptyContainerDir();
+  }
+
+  @VisibleForTesting
+  public void setCheckChunksFilePath(boolean bCheckChunksDirFilePath) {
+    this.bCheckChunksFilePath = bCheckChunksDirFilePath;
   }
 
   @Override

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

@@ -247,12 +247,10 @@ public static void parseKVContainerData(KeyValueContainerData kvContainerData,
     }
     kvContainerData.setDbFile(dbFile);
 
-    boolean bCheckChunksFilePath =
-        config.getBoolean(
-            DatanodeConfiguration.
-              OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE,
-            DatanodeConfiguration.
-              OZONE_DATANODE_CHECK_EMPTY_CONTAINER_ON_DISK_ON_DELETE_DEFAULT);
+    DatanodeConfiguration dnConf =
+        config.getObject(DatanodeConfiguration.class);
+    boolean bCheckChunksFilePath = dnConf.getCheckEmptyContainerDir();
+
     if (kvContainerData.hasSchema(OzoneConsts.SCHEMA_V3)) {
       try (DBHandle db = BlockUtils.getDB(kvContainerData, config)) {
         populateContainerMetadata(kvContainerData,

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteContainerHandler.java

@@ -46,6 +46,7 @@
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
 import org.apache.hadoop.ozone.container.common.interfaces.DBHandle;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
+import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueHandler;
 import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
@@ -138,8 +139,18 @@ public static void shutdown() {
     }
   }
 
+  /**
+   * Delete non-empty container when rocksdb don't have entry about container
+   * but some chunks are left in container directory.
+   * Enable hdds.datanode.check.empty.container.dir.on.delete
+   * Container empty check will return false as some chunks
+   * are left behind in container directory.
+   * Which will ensure container will not be deleted in this case.
+   * @return
+   * @throws IOException
+   */
   @Test(timeout = 60000)
-  public void testDeleteNonEmptyContainerDir()
+  public void testDeleteNonEmptyContainerOnDirEmptyCheckTrue()
       throws Exception {
     // 1. Test if a non force deletion fails if chunks are still present with
     //    block count set to 0
@@ -166,6 +177,10 @@ public void testDeleteNonEmptyContainerDir()
 
     DatanodeDetails datanodeDetails = hddsDatanodeService.getDatanodeDetails();
 
+    KeyValueContainer kv = (KeyValueContainer) getContainerfromDN(
+        hddsDatanodeService, containerId.getId());
+    kv.setCheckChunksFilePath(true);
+
     NodeManager nodeManager =
         cluster.getStorageContainerManager().getScmNodeManager();
     //send the order to close the container
@@ -256,6 +271,101 @@ public void testDeleteNonEmptyContainerDir()
         containerId.getId()));
     Assert.assertTrue(beforeForceCount <
         metrics.getContainerForceDelete());
+
+    kv.setCheckChunksFilePath(false);
+  }
+
+  /**
+   * Delete non-empty container when rocksdb don't have entry about container
+   * but some chunks are left in container directory.
+   * By default, hdds.datanode.check.empty.container.dir.on.delete is false.
+   * Even though chunks are left in container directory,
+   * container empty check will return true
+   * as rocksdb don't have container information.
+   * Container deletion should succeed in this case.
+   * @return
+   * @throws IOException
+   */
+  @Test(timeout = 60000)
+  public void testDeleteNonEmptyContainerOnDirEmptyCheckFalse()
+      throws Exception {
+    //the easiest way to create an open container is creating a key
+    String keyName = UUID.randomUUID().toString();
+
+    // create key
+    createKey(keyName);
+
+    // get containerID of the key
+    ContainerID containerId = getContainerID(keyName);
+
+    // We need to close the container because delete container only happens
+    // on closed containers when force flag is set to false.
+
+    HddsDatanodeService hddsDatanodeService =
+        cluster.getHddsDatanodes().get(0);
+
+    Assert.assertFalse(isContainerClosed(hddsDatanodeService,
+        containerId.getId()));
+
+    DatanodeDetails datanodeDetails = hddsDatanodeService.getDatanodeDetails();
+
+    NodeManager nodeManager =
+        cluster.getStorageContainerManager().getScmNodeManager();
+    //send the order to close the container
+    OzoneTestUtils.closeAllContainers(cluster.getStorageContainerManager()
+        .getEventQueue(), cluster.getStorageContainerManager());
+
+    GenericTestUtils.waitFor(() ->
+            isContainerClosed(hddsDatanodeService, containerId.getId()),
+        500, 5 * 1000);
+
+    //double check if it's really closed (waitFor also throws an exception)
+    Assert.assertTrue(isContainerClosed(hddsDatanodeService,
+        containerId.getId()));
+
+    // Delete key, which will make isEmpty flag to true in containerData
+    objectStore.getVolume(volumeName)
+        .getBucket(bucketName).deleteKey(keyName);
+
+    // Ensure isEmpty flag is true when key is deleted and container is empty
+    GenericTestUtils.waitFor(() -> getContainerfromDN(
+            hddsDatanodeService, containerId.getId())
+            .getContainerData().isEmpty(),
+        500,
+        5 * 2000);
+
+    Container containerInternalObj =
+        hddsDatanodeService.
+            getDatanodeStateMachine().
+            getContainer().getContainerSet().getContainer(containerId.getId());
+
+    // Write a file to the container chunks directory indicating that there
+    // might be a discrepancy between block count as recorded in RocksDB and
+    // what is actually on disk.
+    File lingeringBlock =
+        new File(containerInternalObj.
+            getContainerData().getChunksPath() + "/1.block");
+    lingeringBlock.createNewFile();
+
+    // Check container exists before sending delete container command
+    Assert.assertFalse(isContainerDeleted(hddsDatanodeService,
+        containerId.getId()));
+
+    // send delete container to the datanode
+    SCMCommand<?> command = new DeleteContainerCommand(containerId.getId(),
+        false);
+
+    // Send the delete command. It should succeed as even though
+    // there is a lingering block on disk but isEmpty container flag is true.
+    command.setTerm(
+        cluster.getStorageContainerManager().getScmContext().getTermOfLeader());
+    nodeManager.addDatanodeCommand(datanodeDetails.getUuid(), command);
+
+    GenericTestUtils.waitFor(() ->
+            isContainerDeleted(hddsDatanodeService, containerId.getId()),
+        500, 5 * 1000);
+    Assert.assertTrue(isContainerDeleted(hddsDatanodeService,
+        containerId.getId()));
   }
 
   @Test(timeout = 60000)