HDDS-4426. SCM should create transactions using all blocks received from OM

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManager.java

@@ -21,12 +21,12 @@
 import java.io.IOException;
 import java.util.List;
 
-import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
 import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager.SafeModeStatus;
 import org.apache.hadoop.hdds.server.events.EventHandler;
+import org.apache.hadoop.ozone.common.BlockGroup;
 
 /**
  *
@@ -59,7 +59,7 @@ AllocatedBlock allocateBlock(long size, HddsProtos.ReplicationType type,
    *                 a particular object key.
    * @throws IOException if exception happens, non of the blocks is deleted.
    */
-  void deleteBlocks(List<BlockID> blockIDs) throws IOException;
+  void deleteBlocks(List<BlockGroup> blockIDs) throws IOException;
 
   /**
    * @return the block deletion transaction log maintained by SCM.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java

@@ -50,6 +50,7 @@
 import org.apache.hadoop.hdds.server.events.EventPublisher;
 import org.apache.hadoop.hdds.utils.UniqueId;
 import org.apache.hadoop.metrics2.util.MBeans;
+import org.apache.hadoop.ozone.common.BlockGroup;
 import org.apache.hadoop.util.StringUtils;
 
 import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes.INVALID_BLOCK_SIZE;
@@ -285,38 +286,39 @@ private AllocatedBlock newBlock(ContainerInfo containerInfo) {
    * successful, given blocks are
    * entering pending deletion state and becomes invisible from SCM namespace.
    *
-   * @param blockIDs block IDs. This is often the list of blocks of a
-   * particular object key.
+   * @param keyBlocksInfoList . This is the list of BlockGroup which contains
+   * groupID of keys and list of BlockIDs associated with them.
    * @throws IOException if exception happens, non of the blocks is deleted.
    */
   @Override
-  public void deleteBlocks(List<BlockID> blockIDs) throws IOException {
+  public void deleteBlocks(List<BlockGroup> keyBlocksInfoList)
+      throws IOException {
     ScmUtils.preCheck(ScmOps.deleteBlock, safeModePrecheck);
 
-    LOG.info("Deleting blocks {}", StringUtils.join(",", blockIDs));
     Map<Long, List<Long>> containerBlocks = new HashMap<>();
     // TODO: track the block size info so that we can reclaim the container
     // TODO: used space when the block is deleted.
-    for (BlockID block : blockIDs) {
-      // Merge blocks to a container to blocks mapping,
-      // prepare to persist this info to the deletedBlocksLog.
-      long containerID = block.getContainerID();
-      if (containerBlocks.containsKey(containerID)) {
-        containerBlocks.get(containerID).add(block.getLocalID());
-      } else {
-        List<Long> item = new ArrayList<>();
-        item.add(block.getLocalID());
-        containerBlocks.put(containerID, item);
+    for (BlockGroup bg : keyBlocksInfoList) {
+      LOG.info("Deleting blocks {}",
+          StringUtils.join(",", bg.getBlockIDList()));
+      for (BlockID block : bg.getBlockIDList()) {
+        long containerID = block.getContainerID();
+        if (containerBlocks.containsKey(containerID)) {
+          containerBlocks.get(containerID).add(block.getLocalID());
+        } else {
+          List<Long> item = new ArrayList<>();
+          item.add(block.getLocalID());
+          containerBlocks.put(containerID, item);
+        }
       }
     }
 
     try {
       deletedBlockLog.addTransactions(containerBlocks);
     } catch (IOException e) {
-      throw new IOException(
-          "Skip writing the deleted blocks info to"
-              + " the delLog because addTransaction fails. Batch skipped: "
-              + StringUtils.join(",", blockIDs), e);
+      throw new IOException("Skip writing the deleted blocks info to"
+          + " the delLog because addTransaction fails. " + keyBlocksInfoList
+          .size() + "Keys skipped", e);
     }
     // TODO: Container report handling of the deleted blocks:
     // Remove tombstone and update open container usage.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMBlockProtocolServer.java

@@ -65,6 +65,7 @@
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HANDLER_COUNT_DEFAULT;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes.IO_EXCEPTION;
 import static org.apache.hadoop.hdds.scm.server.StorageContainerManager.startRpcServer;
 import static org.apache.hadoop.hdds.server.ServerUtils.getRemoteUserName;
 import static org.apache.hadoop.hdds.server.ServerUtils.updateRPCListenAddress;
@@ -224,54 +225,46 @@ public List<DeleteBlockGroupResult> deleteKeyBlocks(
     }
     List<DeleteBlockGroupResult> results = new ArrayList<>();
     Map<String, String> auditMap = Maps.newHashMap();
-    for (BlockGroup keyBlocks : keyBlocksInfoList) {
-      ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result resultCode;
-      try {
-        // We delete blocks in an atomic operation to prevent getting
-        // into state like only a partial of blocks are deleted,
-        // which will leave key in an inconsistent state.
-        auditMap.put("keyBlockToDelete", keyBlocks.toString());
-        scm.getScmBlockManager().deleteBlocks(keyBlocks.getBlockIDList());
-        resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
-            .Result.success;
-        AUDIT.logWriteSuccess(
-            buildAuditMessageForSuccess(SCMAction.DELETE_KEY_BLOCK, auditMap)
-        );
-      } catch (SCMException scmEx) {
-        LOG.warn("Fail to delete block: {}", keyBlocks.getGroupID(), scmEx);
-        AUDIT.logWriteFailure(
-            buildAuditMessageForFailure(SCMAction.DELETE_KEY_BLOCK, auditMap,
-                scmEx)
-        );
-        switch (scmEx.getResult()) {
-        case SAFE_MODE_EXCEPTION:
-          resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
-              .Result.safeMode;
-          break;
-        case FAILED_TO_FIND_BLOCK:
-          resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
-              .Result.errorNotFound;
-          break;
-        default:
-          resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
-              .Result.unknownFailure;
-        }
-      } catch (IOException ex) {
-        LOG.warn("Fail to delete blocks for object key: {}", keyBlocks
-            .getGroupID(), ex);
-        AUDIT.logWriteFailure(
-            buildAuditMessageForFailure(SCMAction.DELETE_KEY_BLOCK, auditMap,
-                ex)
-        );
-        resultCode = ScmBlockLocationProtocolProtos.DeleteScmBlockResult
-            .Result.unknownFailure;
+    ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result resultCode;
+    Exception e = null;
+    try {
+      scm.getScmBlockManager().deleteBlocks(keyBlocksInfoList);
+      resultCode = ScmBlockLocationProtocolProtos.
+          DeleteScmBlockResult.Result.success;
+    } catch (IOException ioe) {
+      e = ioe;
+      LOG.warn("Fail to delete {} keys", keyBlocksInfoList.size(), ioe);
+      switch (ioe instanceof SCMException ? ((SCMException) ioe).getResult() :
+          IO_EXCEPTION) {
+      case SAFE_MODE_EXCEPTION:
+        resultCode =
+            ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result.safeMode;
+        break;
+      case FAILED_TO_FIND_BLOCK:
+        resultCode =
+            ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result.
+                errorNotFound;
+        break;
+      default:
+        resultCode =
+            ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result.
+                unknownFailure;
       }
-      List<DeleteBlockResult> blockResultList = new ArrayList<>();
-      for (BlockID blockKey : keyBlocks.getBlockIDList()) {
-        blockResultList.add(new DeleteBlockResult(blockKey, resultCode));
+    }
+    for (BlockGroup bg : keyBlocksInfoList) {
+      auditMap.put("KeyBlockToDelete", bg.toString());
+      List<DeleteBlockResult> blockResult = new ArrayList<>();
+      for (BlockID b : bg.getBlockIDList()) {
+        blockResult.add(new DeleteBlockResult(b, resultCode));
       }
-      results.add(new DeleteBlockGroupResult(keyBlocks.getGroupID(),
-          blockResultList));
+      results.add(new DeleteBlockGroupResult(bg.getGroupID(), blockResult));
+    }
+    if (e == null) {
+      AUDIT.logWriteSuccess(
+          buildAuditMessageForSuccess(SCMAction.DELETE_KEY_BLOCK, auditMap));
+    } else {
+      AUDIT.logWriteFailure(
+          buildAuditMessageForFailure(SCMAction.DELETE_KEY_BLOCK, auditMap, e));
     }
     return results;
   }