HDDS-397. Handle deletion for keys with no blocks. Contributed by Lokesh Jain.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java

@@ -32,6 +32,8 @@
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
 import org.apache.hadoop.ozone.om.helpers.OpenKeySession;
+import org.apache.hadoop.ozone.protocol.proto
+    .OzoneManagerProtocolProtos.KeyLocationList;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyInfo;
 import org.apache.hadoop.util.Time;
 import org.apache.hadoop.utils.BackgroundService;
@@ -447,6 +449,15 @@ public void deleteKey(OmKeyArgs args) throws IOException {
       if (objectValue == null) {
         throw new OMException("Key not found",
             OMException.ResultCodes.FAILED_KEY_NOT_FOUND);
+      } else {
+        // directly delete key with no blocks from db. This key need not be
+        // moved to deleted table.
+        KeyInfo keyInfo = KeyInfo.parseFrom(objectValue);
+        if (isKeyEmpty(keyInfo)) {
+          metadataManager.getKeyTable().delete(objectKey);
+          LOG.debug("Key {} deleted from OM DB", keyName);
+          return;
+        }
       }
       metadataManager.getStore().move(objectKey,
           metadataManager.getKeyTable(),
@@ -463,6 +474,15 @@ public void deleteKey(OmKeyArgs args) throws IOException {
     }
   }
 
+  private boolean isKeyEmpty(KeyInfo keyInfo) {
+    for (KeyLocationList keyLocationList : keyInfo.getKeyLocationListList()) {
+      if (keyLocationList.getKeyLocationsCount() != 0) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   @Override
   public List<OmKeyInfo> listKeys(String volumeName, String bucketName,
       String startKey, String keyPrefix,

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java

@@ -585,14 +585,6 @@ public List<BlockGroup> getPendingDeletionKeys(final int keyCount)
               OmKeyInfo.getFromProtobuf(KeyInfo.parseFrom(kv.getValue()));
           // Get block keys as a list.
           OmKeyLocationInfoGroup latest = info.getLatestVersionLocations();
-          if (latest == null) {
-            // This means that we have a key without any blocks.
-            // BUG-BUG: if this happens the key will never be deleted.
-            // TODO: Right thing to do is to remove this key right here.
-            LOG.warn("Found a key without blocks: {}, skipping for now.",
-                DFSUtil.bytes2String(kv.getKey()));
-            continue;
-          }
           List<BlockID> item = latest.getLocationList().stream()
               .map(b -> new BlockID(b.getContainerID(), b.getLocalID()))
               .collect(Collectors.toList());

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestKeyDeletingService.java

@@ -33,6 +33,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -91,14 +92,18 @@ public void checkIfDeleteServiceisDeletingKeys()
             new ScmBlockLocationTestIngClient(null, null, 0),
             metaMgr, conf, UUID.randomUUID().toString());
     final int keyCount = 100;
-    createAndDeleteKeys(keyManager, keyCount);
+    createAndDeleteKeys(keyManager, keyCount, 1);
     KeyDeletingService keyDeletingService =
         (KeyDeletingService) keyManager.getDeletingService();
     keyManager.start();
+    Assert.assertEquals(
+        keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), keyCount);
     GenericTestUtils.waitFor(
         () -> keyDeletingService.getDeletedKeyCount().get() >= keyCount,
         1000, 10000);
     Assert.assertTrue(keyDeletingService.getRunCount().get() > 1);
+    Assert.assertEquals(
+        keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), 0);
   }
 
   @Test(timeout = 30000)
@@ -112,22 +117,51 @@ public void checkIfDeleteServiceWithFailingSCM()
             new ScmBlockLocationTestIngClient(null, null, 1),
             metaMgr, conf, UUID.randomUUID().toString());
     final int keyCount = 100;
-    createAndDeleteKeys(keyManager, keyCount);
+    createAndDeleteKeys(keyManager, keyCount, 1);
     KeyDeletingService keyDeletingService =
         (KeyDeletingService) keyManager.getDeletingService();
     keyManager.start();
+    Assert.assertEquals(
+        keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), keyCount);
     // Make sure that we have run the background thread 5 times more
     GenericTestUtils.waitFor(
         () -> keyDeletingService.getRunCount().get() >= 5,
         100, 1000);
     // Since SCM calls are failing, deletedKeyCount should be zero.
     Assert.assertEquals(keyDeletingService.getDeletedKeyCount().get(), 0);
-
+    Assert.assertEquals(
-
+        keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), keyCount);
   }
 
-  private void createAndDeleteKeys(KeyManager keyManager, int keyCount)
+  @Test(timeout = 30000)
-      throws IOException {
+  public void checkDeletionForEmptyKey()
+      throws IOException, TimeoutException, InterruptedException {
+    OzoneConfiguration conf = createConfAndInitValues();
+    OmMetadataManagerImpl metaMgr = new OmMetadataManagerImpl(conf);
+    //failCallsFrequency = 1 , means all calls fail.
+    KeyManager keyManager =
+        new KeyManagerImpl(
+            new ScmBlockLocationTestIngClient(null, null, 1),
+            metaMgr, conf, UUID.randomUUID().toString());
+    final int keyCount = 100;
+    createAndDeleteKeys(keyManager, keyCount, 0);
+    KeyDeletingService keyDeletingService =
+        (KeyDeletingService) keyManager.getDeletingService();
+    keyManager.start();
+
+    // Since empty keys are directly deleted from db there should be no
+    // pending deletion keys. Also deletedKeyCount should be zero.
+    Assert.assertEquals(
+        keyManager.getPendingDeletionKeys(Integer.MAX_VALUE).size(), 0);
+    // Make sure that we have run the background thread 2 times or more
+    GenericTestUtils.waitFor(
+        () -> keyDeletingService.getRunCount().get() >= 2,
+        100, 1000);
+    Assert.assertEquals(keyDeletingService.getDeletedKeyCount().get(), 0);
+  }
+
+  private void createAndDeleteKeys(KeyManager keyManager, int keyCount,
+      int numBlocks) throws IOException {
     for (int x = 0; x < keyCount; x++) {
       String volumeName = String.format("volume%s",
           RandomStringUtils.randomAlphanumeric(5));
@@ -153,10 +187,13 @@ private void createAndDeleteKeys(KeyManager keyManager, int keyCount)
               .setVolumeName(volumeName)
               .setBucketName(bucketName)
               .setKeyName(keyName)
+              .setLocationInfoList(new ArrayList<>())
               .build();
       //Open, Commit and Delete the Keys in the Key Manager.
       OpenKeySession session = keyManager.openKey(arg);
-      arg.addLocationInfo(keyManager.allocateBlock(arg, session.getId()));
+      for (int i = 0; i < numBlocks; i++) {
+        arg.addLocationInfo(keyManager.allocateBlock(arg, session.getId()));
+      }
       keyManager.commitKey(arg, session.getId());
       keyManager.deleteKey(arg);
     }