HDDS-10572. Implement multiDelete using OMKeysDeleteRequest

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java

@@ -675,6 +675,16 @@ public void deleteKeys(List<String> keyList) throws IOException {
     proxy.deleteKeys(volumeName, name, keyList);
   }
 
+  /**
+   * Deletes the given list of keys from the bucket.
+   * @param keyList List of the key name to be deleted.
+   * @param isQuiet flag to not throw exception if delete fails
+   * @throws IOException
+   */
+  public Map<String, Pair<String, String>> deleteKeysQuiet(List<String> keyList, Boolean isQuiet) throws IOException {
+    return proxy.deleteKeysQuiet(volumeName, name, keyList, isQuiet);
+  }
+
   /**
    * Rename the keyname from fromKeyName to toKeyName.
    * @param fromKeyName The original key name.

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -24,6 +24,7 @@
 import java.util.Map;
 
 import jakarta.annotation.Nonnull;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicationFactor;
@@ -436,6 +437,18 @@ void deleteKeys(String volumeName, String bucketName,
                   List<String> keyNameList)
       throws IOException;
 
+  /**
+   * Deletes keys through the list.
+   * @param volumeName Name of the Volume
+   * @param bucketName Name of the Bucket
+   * @param keyNameList List of the Key
+   * @param isQuiet flag to not throw exception if delete fails
+   * @throws IOException
+   */
+  Map<String, Pair<String, String>> deleteKeysQuiet(String volumeName, String bucketName,
+                                                    List<String> keyNameList, Boolean isQuiet)
+      throws IOException;
+
   /**
    * Renames an existing key within a bucket.
    * @param volumeName Name of the Volume

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -29,6 +29,7 @@
 import javax.crypto.Cipher;
 import javax.crypto.CipherInputStream;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.crypto.CryptoInputStream;
 import org.apache.hadoop.crypto.CryptoOutputStream;
 import org.apache.hadoop.crypto.key.KeyProvider;
@@ -1642,6 +1643,18 @@ public void deleteKeys(
     ozoneManagerClient.deleteKeys(omDeleteKeys);
   }
 
+  @Override
+  public Map<String, Pair<String, String>> deleteKeysQuiet(
+      String volumeName, String bucketName, List<String> keyNameList, Boolean isQuiet)
+      throws IOException {
+    verifyVolumeName(volumeName);
+    verifyBucketName(bucketName);
+    Preconditions.checkNotNull(keyNameList);
+    OmDeleteKeys omDeleteKeys = new OmDeleteKeys(volumeName, bucketName,
+        keyNameList);
+    return ozoneManagerClient.deleteKeysQuiet(omDeleteKeys, isQuiet);
+  }
+
   @Override
   public void renameKey(String volumeName, String bucketName,
       String fromKeyName, String toKeyName) throws IOException {

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -21,9 +21,11 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 
 import jakarta.annotation.Nonnull;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.fs.SafeModeAction;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList;
 import org.apache.hadoop.ozone.OzoneAcl;
@@ -359,6 +361,21 @@ default void deleteKeys(OmDeleteKeys deleteKeys) throws IOException {
         "this to be implemented, as write requests use a new approach.");
   }
 
+  /**
+   * Deletes existing key/keys. This interface supports delete
+   * multiple keys and a single key. Used by deleting files
+   * through OzoneFileSystem.
+   *
+   * @param deleteKeys
+   * @param isQuiet - flag to not throw exception if delete fails
+   * @throws IOException
+   */
+  default Map<String, Pair<String, String>> deleteKeysQuiet(OmDeleteKeys deleteKeys, Boolean isQuiet)
+      throws IOException {
+    throw new UnsupportedOperationException("OzoneManager does not require " +
+        "this to be implemented, as write requests use a new approach.");
+  }
+
 
   /**
    * Deletes an existing empty bucket from volume.

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -28,6 +29,7 @@
 import com.google.protobuf.Proto2Utils;
 import jakarta.annotation.Nonnull;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.fs.SafeModeAction;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
@@ -953,6 +955,12 @@ public void deleteKey(OmKeyArgs args) throws IOException {
    */
   @Override
   public void deleteKeys(OmDeleteKeys deleteKeys) throws IOException {
+    deleteKeysQuiet(deleteKeys, false);
+  }
+
+  @Override
+  public Map<String, Pair<String, String>> deleteKeysQuiet(OmDeleteKeys deleteKeys, Boolean isQuiet)
+      throws IOException {
     DeleteKeysRequest.Builder req = DeleteKeysRequest.newBuilder();
     DeleteKeyArgs deletedKeys = DeleteKeyArgs.newBuilder()
         .setBucketName(deleteKeys.getBucket())
@@ -962,9 +970,18 @@ public void deleteKeys(OmDeleteKeys deleteKeys) throws IOException {
     OMRequest omRequest = createOMRequest(Type.DeleteKeys)
         .setDeleteKeysRequest(req)
         .build();
+    OMResponse omResponse = submitRequest(omRequest);
 
-    handleError(submitRequest(omRequest));
-
+    if (!isQuiet) {
+      handleError(omResponse);
+    }
+    List<OzoneManagerProtocolProtos.DeleteKeyErrors> deleteKeyErrors =
+        omResponse.getDeleteKeysResponse().getDeleteKeyErrorsList();
+    Map<String, Pair<String, String>> keyToErrors = new HashMap<>();
+    for (OzoneManagerProtocolProtos.DeleteKeyErrors deleteKeyError : deleteKeyErrors) {
+      keyToErrors.put(deleteKeyError.getKey(), Pair.of(deleteKeyError.getErrorCode(), deleteKeyError.getErrorMsg()));
+    }
+    return keyToErrors;
   }
 
   /**

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -1301,9 +1301,16 @@ message DeleteKeyArgs {
     repeated string keys = 3;
 }
 
+message DeleteKeyErrors {
+    required string key = 1;
+    required string errorCode = 2;
+    required string errorMsg = 3;
+}
+
 message DeleteKeysResponse {
     optional DeleteKeyArgs unDeletedKeys = 1;
     optional bool status = 2;
+    repeated DeleteKeyErrors deleteKeyErrors = 3;
 }
 
 message DeleteKeyResponse {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysDeleteRequest.java

@@ -57,6 +57,7 @@
 import java.io.IOException;
 import java.nio.file.InvalidPathException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -95,6 +96,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
     Exception exception = null;
     OMClientResponse omClientResponse = null;
     Result result = null;
+    Map<String, Pair<String, String>> keyToError = new HashMap<>();
 
     OMMetrics omMetrics = ozoneManager.getMetrics();
     omMetrics.incNumKeyDeletes();
@@ -150,6 +152,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
               objectKey);
           deleteKeys.remove(keyName);
           unDeletedKeys.addKeys(keyName);
+          keyToError.put(keyName, Pair.of(OMException.ResultCodes.KEY_NOT_FOUND.name(), "Key does not exist"));
           continue;
         }
 
@@ -167,6 +170,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
           LOG.error("Acl check failed for Key: {}", objectKey, ex);
           deleteKeys.remove(keyName);
           unDeletedKeys.addKeys(keyName);
+          keyToError.put(keyName, Pair.of(OMException.ResultCodes.ACCESS_DENIED.name(), "ACL check failed"));
         }
       }
 
@@ -185,7 +189,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
       final long volumeId = omMetadataManager.getVolumeId(volumeName);
       omClientResponse =
           getOmClientResponse(ozoneManager, omKeyInfoList, dirList, omResponse,
-              unDeletedKeys, deleteStatus, omBucketInfo, volumeId, dbOpenKeys);
+              unDeletedKeys, keyToError, deleteStatus, omBucketInfo, volumeId, dbOpenKeys);
 
       result = Result.SUCCESS;
 
@@ -199,6 +203,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
       // Add all keys which are failed due to any other exception .
       for (int i = indexFailed; i < length; i++) {
         unDeletedKeys.addKeys(deleteKeyArgs.getKeys(i));
+        keyToError.put(deleteKeyArgs.getKeys(i), Pair.of(OMException.ResultCodes.INTERNAL_ERROR.name(),
+            ex.getMessage()));
       }
 
       omResponse.setDeleteKeysResponse(
@@ -260,12 +266,18 @@ protected OMClientResponse getOmClientResponse(OzoneManager ozoneManager,
       List<OmKeyInfo> omKeyInfoList, List<OmKeyInfo> dirList,
       OMResponse.Builder omResponse,
       OzoneManagerProtocolProtos.DeleteKeyArgs.Builder unDeletedKeys,
+      Map<String, Pair<String, String>> keyToErrors,
       boolean deleteStatus, OmBucketInfo omBucketInfo, long volumeId, List<String> dbOpenKeys) {
     OMClientResponse omClientResponse;
+    List<OzoneManagerProtocolProtos.DeleteKeyErrors> deleteKeyErrors = new ArrayList<>();
+    for (Map.Entry<String, Pair<String, String>>  key : keyToErrors.entrySet()) {
+      deleteKeyErrors.add(OzoneManagerProtocolProtos.DeleteKeyErrors.newBuilder()
+          .setKey(key.getKey()).setErrorCode(key.getValue().getLeft()).setErrorMsg(key.getValue().getRight()).build());
+    }
     omClientResponse = new OMKeysDeleteResponse(omResponse
         .setDeleteKeysResponse(
             DeleteKeysResponse.newBuilder().setStatus(deleteStatus)
-                .setUnDeletedKeys(unDeletedKeys))
+                .setUnDeletedKeys(unDeletedKeys).addAllDeleteKeyErrors(deleteKeyErrors))
         .setStatus(deleteStatus ? OK : PARTIAL_DELETE).setSuccess(deleteStatus)
         .build(), omKeyInfoList, ozoneManager.isRatisEnabled(),
         omBucketInfo.copyObject(), dbOpenKeys);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OmKeysDeleteRequestWithFSO.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.ozone.om.request.key;
 
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
 import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
@@ -36,7 +37,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.OK;
 import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.PARTIAL_DELETE;
@@ -146,12 +149,18 @@ protected OMClientResponse getOmClientResponse(OzoneManager ozoneManager,
       List<OmKeyInfo> omKeyInfoList, List<OmKeyInfo> dirList,
       OzoneManagerProtocolProtos.OMResponse.Builder omResponse,
       OzoneManagerProtocolProtos.DeleteKeyArgs.Builder unDeletedKeys,
+      Map<String, Pair<String, String>> keyToErrors,
       boolean deleteStatus, OmBucketInfo omBucketInfo, long volumeId, List<String> dbOpenKeys) {
     OMClientResponse omClientResponse;
+    List<OzoneManagerProtocolProtos.DeleteKeyErrors> deleteKeyErrors = new ArrayList<>();
+    for (Map.Entry<String, Pair<String, String>>  key : keyToErrors.entrySet()) {
+      deleteKeyErrors.add(OzoneManagerProtocolProtos.DeleteKeyErrors.newBuilder()
+          .setKey(key.getKey()).setErrorCode(key.getValue().getLeft()).setErrorMsg(key.getValue().getRight()).build());
+    }
     omClientResponse = new OMKeysDeleteResponseWithFSO(omResponse
         .setDeleteKeysResponse(
             OzoneManagerProtocolProtos.DeleteKeysResponse.newBuilder()
-                .setStatus(deleteStatus).setUnDeletedKeys(unDeletedKeys))
+                .setStatus(deleteStatus).setUnDeletedKeys(unDeletedKeys).addAllDeleteKeyErrors(deleteKeyErrors))
         .setStatus(deleteStatus ? OK : PARTIAL_DELETE).setSuccess(deleteStatus)
         .build(), omKeyInfoList, dirList, ozoneManager.isRatisEnabled(),
         omBucketInfo.copyObject(), volumeId, dbOpenKeys);

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeysDeleteRequest.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteKeyArgs;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteKeyErrors;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteKeysRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.junit.jupiter.api.Test;
@@ -73,6 +74,9 @@ protected void checkDeleteKeysResponse(
         omClientResponse.getOMResponse().getDeleteKeysResponse()
             .getUnDeletedKeys();
     assertEquals(0, unDeletedKeys.getKeysCount());
+    List<DeleteKeyErrors> keyErrors =  omClientResponse.getOMResponse().getDeleteKeysResponse()
+        .getDeleteKeyErrorsList();
+    assertEquals(0, keyErrors.size());
 
     // Check all keys are deleted.
     for (String deleteKey : deleteKeyList) {
@@ -123,6 +127,9 @@ protected void checkDeleteKeysResponseForFailure(
         .getDeleteKeysResponse().getUnDeletedKeys();
     assertEquals(1,
         unDeletedKeys.getKeysCount());
+    List<DeleteKeyErrors> keyErrors =  omClientResponse.getOMResponse().getDeleteKeysResponse()
+        .getDeleteKeyErrorsList();
+    assertEquals(1, keyErrors.size());
     assertEquals("dummy", unDeletedKeys.getKeys(0));
   }