diff --git a/constants.js b/constants.js
index 5f792fbf0d..6ea8fdd445 100644
--- a/constants.js
+++ b/constants.js
@@ -205,6 +205,7 @@ const constants = {
         NON_CURRENT_TYPE: 'noncurrent',
         ORPHAN_DM_TYPE: 'orphan',
     },
+    multiObjectDeleteConcurrency: 50,
 };
 
 module.exports = constants;
diff --git a/lib/Config.js b/lib/Config.js
index 778b53e9bc..d090090fa5 100644
--- a/lib/Config.js
+++ b/lib/Config.js
@@ -19,6 +19,7 @@ const { azureAccountNameRegex, base64Regex,
 const { utapiVersion } = require('utapi');
 const { scaleMsPerDay } = s3middleware.objectUtils;
 
+const constants = require('../constants');
 
 // config paths
 const configSearchPaths = [
@@ -1622,6 +1623,16 @@ class Config extends EventEmitter {
         this.overlayVersion = config.overlayVersion || 0;
 
         this._setTimeOptions();
+        this.multiObjectDeleteConcurrency = constants.multiObjectDeleteConcurrency;
+        const extractedNumber = Number.parseInt(config.multiObjectDeleteConcurrency, 10);
+        if (!isNaN(extractedNumber) && extractedNumber > 0 && extractedNumber < 1000) {
+            this.multiObjectDeleteConcurrency = extractedNumber;
+        }
+
+        this.multiObjectDeleteEnableOptimizations = true;
+        if (config.multiObjectDeleteEnableOptimizations === false) {
+            this.multiObjectDeleteEnableOptimizations = false;
+        }
     }
 
     _setTimeOptions() {
diff --git a/lib/api/apiUtils/object/deleteObject.js b/lib/api/apiUtils/object/deleteObject.js
new file mode 100644
index 0000000000..214c5f2bc2
--- /dev/null
+++ b/lib/api/apiUtils/object/deleteObject.js
@@ -0,0 +1,18 @@
+/**
+ * _bucketRequiresOplogUpdate - DELETE an object from a bucket
+ * @param {BucketInfo} bucket - bucket object
+ * @return {boolean} whether objects require oplog updates on deletion, or not
+ */
+function _bucketRequiresOplogUpdate(bucket) {
+    // Default behavior is to require an oplog update
+    if (!bucket || !bucket.getLifecycleConfiguration || !bucket.getNotificationConfiguration) {
+        return true;
+    }
+    // If the bucket has lifecycle configuration or notification configuration
+    // set, we also require an oplog update
+    return bucket.getLifecycleConfiguration() || bucket.getNotificationConfiguration();
+}
+
+module.exports = {
+    _bucketRequiresOplogUpdate,
+};
diff --git a/lib/api/multiObjectDelete.js b/lib/api/multiObjectDelete.js
index aec0e90c04..8d1a0fb25e 100644
--- a/lib/api/multiObjectDelete.js
+++ b/lib/api/multiObjectDelete.js
@@ -17,7 +17,7 @@ const { preprocessingVersioningDelete }
     = require('./apiUtils/object/versioning');
 const createAndStoreObject = require('./apiUtils/object/createAndStoreObject');
 const monitoring = require('../utilities/monitoringHandler');
-const { metadataGetObject } = require('../metadata/metadataUtils');
+const metadataUtils = require('../metadata/metadataUtils');
 const { config } = require('../Config');
 const { isRequesterNonAccountUser } = require('./apiUtils/authorization/permissionChecks');
 const { hasGovernanceBypassHeader, checkUserGovernanceBypass, ObjectLockInfo }
@@ -25,9 +25,11 @@ const { hasGovernanceBypassHeader, checkUserGovernanceBypass, ObjectLockInfo }
 const requestUtils = policies.requestUtils;
 const { validObjectKeys } = require('../routes/routeVeeam');
 const { deleteVeeamCapabilities } = require('../routes/veeam/delete');
+const { _bucketRequiresOplogUpdate } = require('./apiUtils/object/deleteObject');
 
 const versionIdUtils = versioning.VersionID;
-
+const { data } = require('../data/wrapper');
+const logger = require('../utilities/logger');
 
 /*
    Format of xml request:
@@ -169,6 +171,63 @@ function _parseXml(xmlToParse, next) {
     });
 }
 
+/**
+ * decodeObjectVersion - decode object version to be deleted
+ * @param {object} entry - entry from data model
+ * @param {function} next - callback to call with error or decoded version
+ * @return {undefined}
+ **/
+function decodeObjectVersion(entry) {
+    let decodedVersionId;
+    if (entry.versionId) {
+        decodedVersionId = entry.versionId === 'null' ?
+            'null' : versionIdUtils.decode(entry.versionId);
+    }
+    if (decodedVersionId instanceof Error) {
+        return [errors.NoSuchVersion];
+    }
+    return [null, decodedVersionId];
+}
+
+/**
+ * Initialization function for the MultiObjectDelete API that will, based on the
+ * current metadata backend, assess if metadata READ batching is supported. If
+ * yes, the initialization step will call the metadataGetObjects function from
+ * the MetadataWrapper.
+ * @param {string} bucketName - bucket name
+ * @param {string []} inPlay - list of object keys still in play
+ * @param {object} log - logger object
+ * @param {function} callback - callback to call with error or list of objects
+ * @return {undefined}
+ */
+function initializeMultiObjectDeleteWithBatchingSupport(bucketName, inPlay, log, callback) {
+    if (config.multiObjectDeleteEnableOptimizations === false) {
+        return callback(null, {});
+    }
+    // If the backend supports batching, we want to optimize the API latency by
+    // first getting all the objects metadata, stored in memory, for later use
+    // in the API. This approach does not change the API architecture, but
+    // transplants an additional piece of code that can greatly improve the API
+    // latency when the database supports batching.
+    const objectKeys = Object.values(inPlay).map(entry => {
+        const [err, versionId] = decodeObjectVersion(entry, bucketName);
+        if (err) {
+            return null;
+        }
+        return {
+            versionId,
+            inPlay: entry,
+        };
+    });
+    return metadataUtils.metadataGetObjects(bucketName, objectKeys, log, (err, cache) => {
+        // This optional step is read-only, so any error can be safely ignored
+        if (err) {
+            return callback(null, {});
+        }
+        return callback(null, cache);
+    });
+}
+
 /**
 * gets object metadata and deletes object
 * @param {AuthInfo} authInfo - Instance of AuthInfo class with requester's info
@@ -194,34 +253,18 @@ function getObjMetadataAndDelete(authInfo, canonicalID, request,
     let numOfObjectsRemoved = 0;
     const skipError = new Error('skip');
     const objectLockedError = new Error('object locked');
+    let deleteFromStorage = [];
 
-    // doing 5 requests at a time. note that the data wrapper
-    // will do 5 parallel requests to data backend to delete parts
-    return async.forEachLimit(inPlay, 5, (entry, moveOn) => {
-        async.waterfall([
-            callback => {
-                let decodedVersionId;
-                if (entry.versionId) {
-                    decodedVersionId = entry.versionId === 'null' ?
-                        'null' : versionIdUtils.decode(entry.versionId);
-                }
-                if (decodedVersionId instanceof Error) {
-                    monitoring.promMetrics('DELETE', bucketName, 404,
-                        'multiObjectDelete');
-                    return callback(errors.NoSuchVersion);
-                }
-                return callback(null, decodedVersionId);
-            },
-            // for obj deletes, no need to check acl's at object level
-            // (authority is at the bucket level for obj deletes)
-            (versionId, callback) => metadataGetObject(bucketName, entry.key,
-                versionId, log, (err, objMD) => {
-                    // if general error from metadata return error
-                    if (err) {
-                        monitoring.promMetrics('DELETE', bucketName, err.code,
-                            'multiObjectDelete');
-                        return callback(err);
-                    }
+    return async.waterfall([
+        callback => initializeMultiObjectDeleteWithBatchingSupport(bucketName, inPlay, log, callback),
+        (cache, callback) => async.forEachLimit(inPlay, config.multiObjectDeleteConcurrency, (entry, moveOn) => {
+            async.waterfall([
+                callback => callback(...decodeObjectVersion(entry, bucketName)),
+                // for obj deletes, no need to check acl's at object level
+                // (authority is at the bucket level for obj deletes)
+                (versionId, callback) => metadataUtils.metadataGetObject(bucketName, entry.key,
+                    versionId, cache, log, (err, objMD) => callback(err, objMD, versionId)),
+                (objMD, versionId, callback) => {
                     if (!objMD) {
                         const verCfg = bucket.getVersioningConfiguration();
                         // To adhere to AWS behavior, create a delete marker
@@ -229,7 +272,7 @@ function getObjMetadataAndDelete(authInfo, canonicalID, request,
                         // when versioning has been configured
                         if (verCfg && !entry.versionId) {
                             log.debug('trying to delete specific version ' +
-                            ' that does not exist');
+                                'that does not exist');
                             return callback(null, objMD, versionId);
                         }
                         // otherwise if particular key does not exist, AWS
@@ -245,114 +288,160 @@ function getObjMetadataAndDelete(authInfo, canonicalID, request,
                         objMD.location[0].deleteVersion = true;
                     }
                     return callback(null, objMD, versionId);
-                }),
-            (objMD, versionId, callback) => {
-                // AWS only returns an object lock error if a version id
-                // is specified, else continue to create a delete marker
-                if (!versionId || !bucket.isObjectLockEnabled()) {
-                    return callback(null, null, objMD, versionId);
-                }
-                const hasGovernanceBypass = hasGovernanceBypassHeader(request.headers);
-                if (hasGovernanceBypass && isRequesterNonAccountUser(authInfo)) {
-                    return checkUserGovernanceBypass(request, authInfo, bucket, entry.key, log, error => {
-                        if (error && error.is.AccessDenied) {
-                            log.debug('user does not have BypassGovernanceRetention and object is locked', { error });
-                            return callback(objectLockedError);
+                },
+                (objMD, versionId, callback) => {
+                    // AWS only returns an object lock error if a version id
+                    // is specified, else continue to create a delete marker
+                    if (!versionId || !bucket.isObjectLockEnabled()) {
+                        return callback(null, null, objMD, versionId);
+                    }
+                    const hasGovernanceBypass = hasGovernanceBypassHeader(request.headers);
+                    if (hasGovernanceBypass && isRequesterNonAccountUser(authInfo)) {
+                        return checkUserGovernanceBypass(request, authInfo, bucket, entry.key, log, error => {
+                            if (error && error.is.AccessDenied) {
+                                log.debug('user does not have BypassGovernanceRetention and object is locked',
+                                    { error });
+                                return callback(objectLockedError);
+                            }
+                            if (error) {
+                                return callback(error);
+                            }
+                            return callback(null, hasGovernanceBypass, objMD, versionId);
+                        });
+                    }
+                    return callback(null, hasGovernanceBypass, objMD, versionId);
+                },
+                (hasGovernanceBypass, objMD, versionId, callback) => {
+                    // AWS only returns an object lock error if a version id
+                    // is specified, else continue to create a delete marker
+                    if (!versionId || !bucket.isObjectLockEnabled()) {
+                        return callback(null, objMD, versionId);
+                    }
+                    const objLockInfo = new ObjectLockInfo({
+                        mode: objMD.retentionMode,
+                        date: objMD.retentionDate,
+                        legalHold: objMD.legalHold || false,
+                    });
+
+                    // If the object can not be deleted raise an error
+                    if (!objLockInfo.canModifyObject(hasGovernanceBypass)) {
+                        log.debug('trying to delete locked object');
+                        return callback(objectLockedError);
+                    }
+
+                    return callback(null, objMD, versionId);
+                },
+                (objMD, versionId, callback) => {
+                    const options = preprocessingVersioningDelete(
+                        bucketName, bucket, objMD, versionId, config.nullVersionCompatMode);
+                    const deleteInfo = {};
+                    if (options && options.deleteData) {
+                        deleteInfo.deleted = true;
+                        if (!_bucketRequiresOplogUpdate(bucket)) {
+                            options.doesNotNeedOpogUpdate = true;
                         }
-                        if (error) {
-                            return callback(error);
+                        if (objMD.uploadId) {
+                            // eslint-disable-next-line
+                            options.replayId = objMD.uploadId;
                         }
-                        return callback(null, hasGovernanceBypass, objMD, versionId);
-                    });
+                        return services.deleteObject(bucketName, objMD,
+                            entry.key, options, config.multiObjectDeleteEnableOptimizations, log,
+                            's3:ObjectRemoved:Delete', (err, toDelete) => {
+                                if (err) {
+                                    return callback(err);
+                                }
+                                if (toDelete) {
+                                    deleteFromStorage = deleteFromStorage.concat(toDelete);
+                                }
+                                return callback(null, objMD, deleteInfo);
+                            });
+                    }
+                    deleteInfo.newDeleteMarker = true;
+                    // This call will create a delete-marker
+                    return createAndStoreObject(bucketName, bucket, entry.key,
+                        objMD, authInfo, canonicalID, null, request,
+                        deleteInfo.newDeleteMarker, null, log, 's3:ObjectRemoved:DeleteMarkerCreated', (err, result) =>
+                        callback(err, objMD, deleteInfo, result.versionId));
+                },
+            ], (err, objMD, deleteInfo, versionId) => {
+                if (err === skipError) {
+                    return moveOn();
+                } else if (err === objectLockedError) {
+                    errorResults.push({ entry, error: errors.AccessDenied, objectLocked: true });
+                    return moveOn();
+                } else if (err) {
+                    log.error('error deleting object', { error: err, entry });
+                    errorResults.push({ entry, error: err });
+                    return moveOn();
                 }
-                return callback(null, hasGovernanceBypass, objMD, versionId);
-            },
-            (hasGovernanceBypass, objMD, versionId, callback) => {
-                // AWS only returns an object lock error if a version id
-                // is specified, else continue to create a delete marker
-                if (!versionId || !bucket.isObjectLockEnabled()) {
-                    return callback(null, objMD, versionId);
+                if (deleteInfo.deleted && objMD['content-length']) {
+                    numOfObjectsRemoved++;
+                    totalContentLengthDeleted += objMD['content-length'];
                 }
-                const objLockInfo = new ObjectLockInfo({
-                    mode: objMD.retentionMode,
-                    date: objMD.retentionDate,
-                    legalHold: objMD.legalHold || false,
+                let isDeleteMarker;
+                let deleteMarkerVersionId;
+                // - If trying to delete an object that does not exist (if a new
+                // delete marker was created)
+                // - Or if an object exists but no version was specified
+                // return DeleteMarkerVersionId equals the versionID of the marker
+                // you just generated and DeleteMarker tag equals true
+                if (deleteInfo.newDeleteMarker) {
+                    isDeleteMarker = true;
+                    deleteMarkerVersionId = versionIdUtils.encode(versionId);
+                    // In this case we are putting a new object (i.e., the delete
+                    // marker), so we decrement the numOfObjectsRemoved value.
+                    numOfObjectsRemoved--;
+                    // If trying to delete a delete marker, DeleteMarkerVersionId equals
+                    // deleteMarker's versionID and DeleteMarker equals true
+                } else if (objMD && objMD.isDeleteMarker) {
+                    isDeleteMarker = true;
+                    deleteMarkerVersionId = entry.versionId;
+                }
+                successfullyDeleted.push({
+                    entry, isDeleteMarker,
+                    deleteMarkerVersionId,
                 });
+                return moveOn();
+            });
+        },
+            // end of forEach func
+            err => {
+                // Batch delete all objects
+                const onDone = () => callback(err, quietSetting, errorResults, numOfObjectsRemoved,
+                    successfullyDeleted, totalContentLengthDeleted, bucket);
 
-                // If the object can not be deleted raise an error
-                if (!objLockInfo.canModifyObject(hasGovernanceBypass)) {
-                    log.debug('trying to delete locked object');
-                    return callback(objectLockedError);
+                if (err && deleteFromStorage.length === 0) {
+                    log.trace('no objects to delete from data backend');
+                    return onDone();
                 }
+                // If error but we have objects in the list, delete them to ensure
+                // consistent state.
+                log.trace('deleting objects from data backend');
 
-                return callback(null, objMD, versionId);
-            },
-            (objMD, versionId, callback) => {
-                const options = preprocessingVersioningDelete(
-                    bucketName, bucket, objMD, versionId, config.nullVersionCompatMode);
-                const deleteInfo = {};
-                if (options && options.deleteData) {
-                    deleteInfo.deleted = true;
-                    if (objMD.uploadId) {
-                        // eslint-disable-next-line
-                        options.replayId = objMD.uploadId;
-                    }
-                    return services.deleteObject(bucketName, objMD,
-                        entry.key, options, log,  's3:ObjectRemoved:Delete', err =>
-                        callback(err, objMD, deleteInfo));
+                // Split the array into chunks
+                const chunks = [];
+                while (deleteFromStorage.length > 0) {
+                    chunks.push(deleteFromStorage.splice(0, config.multiObjectDeleteConcurrency));
                 }
-                deleteInfo.newDeleteMarker = true;
-                // This call will create a delete-marker
-                return createAndStoreObject(bucketName, bucket, entry.key,
-                    objMD, authInfo, canonicalID, null, request,
-                    deleteInfo.newDeleteMarker, null, log, 's3:ObjectRemoved:DeleteMarkerCreated',
-                    (err, result) =>
-                    callback(err, objMD, deleteInfo, result.versionId));
-            },
-        ], (err, objMD, deleteInfo, versionId) => {
-            if (err === skipError) {
-                return moveOn();
-            } else if (err === objectLockedError) {
-                errorResults.push({ entry, error: errors.AccessDenied, objectLocked: true });
-                return moveOn();
-            } else if (err) {
-                log.error('error deleting object', { error: err, entry });
-                errorResults.push({ entry, error: err });
-                return moveOn();
-            }
-            if (deleteInfo.deleted && objMD['content-length']) {
-                numOfObjectsRemoved++;
-                totalContentLengthDeleted += objMD['content-length'];
-            }
-            let isDeleteMarker;
-            let deleteMarkerVersionId;
-            // - If trying to delete an object that does not exist (if a new
-            // delete marker was created)
-            // - Or if an object exists but no version was specified
-            // return DeleteMarkerVersionId equals the versionID of the marker
-            // you just generated and DeleteMarker tag equals true
-            if (deleteInfo.newDeleteMarker) {
-                isDeleteMarker = true;
-                deleteMarkerVersionId = versionIdUtils.encode(versionId);
-                // In this case we are putting a new object (i.e., the delete
-                // marker), so we decrement the numOfObjectsRemoved value.
-                numOfObjectsRemoved--;
-            // If trying to delete a delete marker, DeleteMarkerVersionId equals
-            // deleteMarker's versionID and DeleteMarker equals true
-            } else if (objMD && objMD.isDeleteMarker) {
-                isDeleteMarker = true;
-                deleteMarkerVersionId = entry.versionId;
-            }
-            successfullyDeleted.push({ entry, isDeleteMarker,
-                deleteMarkerVersionId });
-            return moveOn();
-        });
-    },
-    // end of forEach func
-    err => {
-        log.trace('finished deleting objects', { numOfObjectsRemoved });
-        return next(err, quietSetting, errorResults, numOfObjectsRemoved,
-            successfullyDeleted, totalContentLengthDeleted, bucket);
+
+                return async.each(chunks, (chunk, done) => data.batchDelete(chunk, null, null,
+                    logger.newRequestLoggerFromSerializedUids(log.getSerializedUids()), done),
+                    err => {
+                        if (err) {
+                            log.error('error deleting objects from data backend', { error: err });
+                            return onDone(err);
+                        }
+                        return onDone();
+                    });
+            }),
+    ], (err, ...results) => {
+        // if general error from metadata return error
+        if (err) {
+            monitoring.promMetrics('DELETE', bucketName, err.code,
+                'multiObjectDelete');
+            return next(err);
+        }
+        return next(null, ...results);
     });
 }
 
@@ -589,4 +678,6 @@ function multiObjectDelete(authInfo, request, log, callback) {
 module.exports = {
     getObjMetadataAndDelete,
     multiObjectDelete,
+    decodeObjectVersion,
+    initializeMultiObjectDeleteWithBatchingSupport,
 };
diff --git a/lib/api/objectDelete.js b/lib/api/objectDelete.js
index fc431fddfb..73e4e3d871 100644
--- a/lib/api/objectDelete.js
+++ b/lib/api/objectDelete.js
@@ -14,6 +14,7 @@ const { hasGovernanceBypassHeader, checkUserGovernanceBypass, ObjectLockInfo }
     = require('./apiUtils/object/objectLockHelpers');
 const { isRequesterNonAccountUser } = require('./apiUtils/authorization/permissionChecks');
 const { config } = require('../Config');
+const { _bucketRequiresOplogUpdate } = require('./apiUtils/object/deleteObject');
 
 const versionIdUtils = versioning.VersionID;
 const objectLockedError = new Error('object locked');
@@ -178,7 +179,7 @@ function objectDeleteInternal(authInfo, request, log, isExpiration, cb) {
                             deleteInfo.removeDeleteMarker = true;
                         }
                         return services.deleteObject(bucketName, objectMD,
-                            objectKey, delOptions, log, isExpiration ?
+                            objectKey, delOptions, false, log, isExpiration ?
                                 's3:LifecycleExpiration:Delete' :
                                 's3:ObjectRemoved:Delete',
                             (err, delResult) =>
@@ -197,8 +198,12 @@ function objectDeleteInternal(authInfo, request, log, isExpiration, cb) {
                     delOptions.replayId = objectMD.uploadId;
                 }
 
+                if (!_bucketRequiresOplogUpdate(bucketMD)) {
+                    delOptions.doesNotNeedOpogUpdate = true;
+                }
+
                 return services.deleteObject(bucketName, objectMD, objectKey,
-                    delOptions, log, isExpiration ?
+                    delOptions, false, log, isExpiration ?
                         's3:LifecycleExpiration:Delete' :
                         's3:ObjectRemoved:Delete',
                     (err, delResult) => next(err, bucketMD,
diff --git a/lib/metadata/metadataUtils.js b/lib/metadata/metadataUtils.js
index d931d5d5c9..ef93f24b23 100644
--- a/lib/metadata/metadataUtils.js
+++ b/lib/metadata/metadataUtils.js
@@ -60,13 +60,18 @@ function getNullVersionFromMaster(bucketName, objectKey, log, cb) {
  * @param {string} bucketName - name of bucket
  * @param {string} objectKey - name of object key
  * @param {string} [versionId] - version of object to retrieve
+ * @param {object} cachedDocuments - cached version of the documents used for
+ *                                   abstraction purposes
  * @param {RequestLogger} log - request logger
  * @param {function} cb - callback
  * @return {undefined} - and call callback with err, bucket md and object md
  */
-function metadataGetObject(bucketName, objectKey, versionId, log, cb) {
+function metadataGetObject(bucketName, objectKey, versionId, cachedDocuments, log, cb) {
     // versionId may be 'null', which asks metadata to fetch the null key specifically
     const options = { versionId, getDeleteMarker: true };
+    if (cachedDocuments && cachedDocuments[objectKey]) {
+        return cb(null, cachedDocuments[objectKey]);
+    }
     return metadata.getObjectMD(bucketName, objectKey, options, log,
         (err, objMD) => {
             if (err) {
@@ -84,6 +89,40 @@ function metadataGetObject(bucketName, objectKey, versionId, log, cb) {
         });
 }
 
+/** metadataGetObjects - retrieves specified object or version from metadata. This
+ * method uses cursors, hence is only compatible with a MongoDB DB backend.
+ * @param {string} bucketName - name of bucket
+ * @param {string} objectsKeys - name of object key
+ * @param {RequestLogger} log - request logger
+ * @param {function} cb - callback
+ * @return {undefined} - and call callback with err, bucket md and object md
+ */
+function metadataGetObjects(bucketName, objectsKeys, log, cb) {
+    const options = { getDeleteMarker: true };
+    const objects = objectsKeys.map(objectKey => ({
+        key: objectKey ? objectKey.inPlay.key : null,
+        params: options,
+        versionId: objectKey ? objectKey.versionId : null,
+    }));
+
+    // Returned objects are following the following format: { key, doc, versionId }
+    // That is required with batching to properly map the objects
+    return metadata.getObjectsMD(bucketName, objects, log, (err, objMds) => {
+        if (err) {
+            log.debug('error getting objects MD from metadata', { error: err });
+            return cb(err);
+        }
+
+        const result = {};
+        objMds.forEach(objMd => {
+            if (objMd.doc) {
+                result[`${objMd.doc.key}${objMd.versionId}`] = objMd.doc;
+            }
+        });
+
+        return cb(null, result);
+    });
+}
 /**
  * Validate that a bucket is accessible and authorized to the user,
  * return a specific error code otherwise
@@ -215,6 +254,7 @@ function metadataValidateBucket(params, log, callback) {
 module.exports = {
     validateBucket,
     metadataGetObject,
+    metadataGetObjects,
     metadataValidateBucketAndObj,
     metadataValidateBucket,
 };
diff --git a/lib/routes/routeBackbeat.js b/lib/routes/routeBackbeat.js
index 87ea796ed7..7f86524f8e 100644
--- a/lib/routes/routeBackbeat.js
+++ b/lib/routes/routeBackbeat.js
@@ -951,7 +951,7 @@ function putObjectTagging(request, response, log, callback) {
     // retrieve it from metadata here.
     if (dataStoreVersionId === '') {
         return metadataGetObject(sourceBucket, request.objectKey,
-            sourceVersionId, log, (err, objMD) => {
+            sourceVersionId, null, log, (err, objMD) => {
                 if (err) {
                     return callback(err);
                 }
@@ -983,7 +983,7 @@ function deleteObjectTagging(request, response, log, callback) {
     // retrieve it from metadata here.
     if (dataStoreVersionId === '') {
         return metadataGetObject(sourceBucket, request.objectKey,
-            sourceVersionId, log, (err, objMD) => {
+            sourceVersionId, null, log, (err, objMD) => {
                 if (err) {
                     return callback(err);
                 }
diff --git a/lib/services.js b/lib/services.js
index eda0528cfa..e685e8542c 100644
--- a/lib/services.js
+++ b/lib/services.js
@@ -306,12 +306,14 @@ const services = {
      * @param {string} objectKey - object key name
      * @param {object} options - other instructions, such as { versionId } to
      *                           delete a specific version of the object
+     * @param {boolean} deferLocationDeletion - true if the object should not
+     * be removed from the storage, but be returned instead.
      * @param {Log} log - logger instance
      * @param {string} originOp - origin operation
      * @param {function} cb - callback from async.waterfall in objectGet
      * @return {undefined}
      */
-    deleteObject(bucketName, objectMD, objectKey, options, log, originOp, cb) {
+    deleteObject(bucketName, objectMD, objectKey, options, deferLocationDeletion, log, originOp, cb) {
         log.trace('deleting object from bucket');
         assert.strictEqual(typeof bucketName, 'string');
         assert.strictEqual(typeof objectMD, 'object');
@@ -328,12 +330,19 @@ const services = {
                             log.getSerializedUids());
                     if (objectMD.location === null) {
                         return cb(null, res);
-                    } else if (!Array.isArray(objectMD.location)) {
+                    }
+
+                    if (deferLocationDeletion) {
+                        return cb(null, Array.isArray(objectMD.location)
+                            ? objectMD.location : [objectMD.location]);
+                    }
+
+                    if (!Array.isArray(objectMD.location)) {
                         data.delete(objectMD.location, deleteLog);
                         return cb(null, res);
                     }
-                    return data.batchDelete(objectMD.location, null, null,
-                    deleteLog, err => {
+
+                    return data.batchDelete(objectMD.location, null, null, deleteLog, err => {
                         if (err) {
                             return cb(err);
                         }
diff --git a/package.json b/package.json
index d6b78815ea..7a3d66cf5a 100644
--- a/package.json
+++ b/package.json
@@ -21,7 +21,7 @@
   "dependencies": {
     "@azure/storage-blob": "^12.12.0",
     "@hapi/joi": "^17.1.0",
-    "arsenal": "git+https://github.com/scality/arsenal#8.1.103",
+    "arsenal": "git+https://github.com/scality/arsenal#8.1.104",
     "async": "~2.5.0",
     "aws-sdk": "2.905.0",
     "bucketclient": "scality/bucketclient#8.1.9",
diff --git a/tests/unit/api/multiObjectDelete.js b/tests/unit/api/multiObjectDelete.js
index f9bc9831de..1f16d311a1 100644
--- a/tests/unit/api/multiObjectDelete.js
+++ b/tests/unit/api/multiObjectDelete.js
@@ -1,17 +1,19 @@
 const assert = require('assert');
 const { errors, storage } = require('arsenal');
 
-const { getObjMetadataAndDelete }
+const { decodeObjectVersion, getObjMetadataAndDelete, initializeMultiObjectDeleteWithBatchingSupport }
     = require('../../../lib/api/multiObjectDelete');
+const multiObjectDelete = require('../../../lib/api/multiObjectDelete');
 const { cleanup, DummyRequestLogger, makeAuthInfo } = require('../helpers');
 const DummyRequest = require('../DummyRequest');
 const { bucketPut } = require('../../../lib/api/bucketPut');
 const objectPut = require('../../../lib/api/objectPut');
+const log = new DummyRequestLogger();
 
 const { metadata } = storage.metadata.inMemory.metadata;
 const { ds } = storage.data.inMemory.datastore;
+const sinon = require('sinon');
 
-const log = new DummyRequestLogger();
 const canonicalID = 'accessKey1';
 const authInfo = makeAuthInfo(canonicalID);
 const namespace = 'default';
@@ -20,6 +22,8 @@ const postBody = Buffer.from('I am a body', 'utf8');
 const contentLength = 2 * postBody.length;
 const objectKey1 = 'objectName1';
 const objectKey2 = 'objectName2';
+const metadataUtils = require('../../../lib/metadata/metadataUtils');
+const services = require('../../../lib/services');
 const testBucketPutRequest = new DummyRequest({
     bucketName,
     namespace,
@@ -69,6 +73,10 @@ describe('getObjMetadataAndDelete function for multiObjectDelete', () => {
         });
     });
 
+    afterEach(() => {
+        sinon.restore();
+    });
+
     it('should successfully get object metadata and then ' +
         'delete metadata and data', done => {
         getObjMetadataAndDelete(authInfo, 'foo', request, bucketName, bucket,
@@ -185,4 +193,139 @@ describe('getObjMetadataAndDelete function for multiObjectDelete', () => {
                 done();
             });
     });
+
+    it('should properly batch delete data even if there are errors in other objects', done => {
+        const deleteObjectStub = sinon.stub(services, 'deleteObject');
+        deleteObjectStub.onCall(0).callsArgWith(7, errors.InternalError);
+        deleteObjectStub.onCall(1).callsArgWith(7, null);
+
+        getObjMetadataAndDelete(authInfo, 'foo', request, bucketName, bucket,
+        true, [], [{ key: objectKey1 }, { key: objectKey2 }], log,
+        (err, quietSetting, errorResults, numOfObjects,
+            successfullyDeleted, totalContentLengthDeleted) => {
+            assert.ifError(err);
+            assert.strictEqual(quietSetting, true);
+            assert.deepStrictEqual(errorResults, [
+                {
+                    entry: {
+                        key: objectKey1,
+                    },
+                    error: errors.InternalError,
+                },
+            ]);
+            assert.strictEqual(numOfObjects, 1);
+            assert.strictEqual(totalContentLengthDeleted, contentLength / 2);
+            // Expect still in memory as we stubbed the function
+            assert.strictEqual(metadata.keyMaps.get(bucketName).has(objectKey1), true);
+            assert.strictEqual(metadata.keyMaps.get(bucketName).has(objectKey2), true);
+            // ensure object 2 only is in the list of successful deletions
+            assert.strictEqual(successfullyDeleted.length, 1);
+            assert.deepStrictEqual(successfullyDeleted[0].entry.key, objectKey2);
+            return done();
+        });
+    });
+});
+
+describe('initializeMultiObjectDeleteWithBatchingSupport', () => {
+    let bucketName;
+    let inPlay;
+    let log;
+    let callback;
+
+    beforeEach(() => {
+        bucketName = 'myBucket';
+        inPlay = { one: 'object1', two: 'object2' };
+        log = {};
+        callback = sinon.spy();
+    });
+
+    afterEach(() => {
+        sinon.restore();
+    });
+
+    it('should not throw if the decodeObjectVersion function fails', done => {
+        const metadataGetObjectsStub = sinon.stub(metadataUtils, 'metadataGetObjects').yields(null, {});
+        sinon.stub(multiObjectDelete, 'decodeObjectVersion').returns([new Error('decode error')]);
+
+        initializeMultiObjectDeleteWithBatchingSupport(bucketName, inPlay, log, callback);
+
+        assert.strictEqual(metadataGetObjectsStub.callCount, 1);
+        sinon.assert.calledOnce(callback);
+        assert.strictEqual(callback.getCall(0).args[0], null);
+        assert.deepStrictEqual(callback.getCall(0).args[1], {});
+        done();
+    });
+
+    it('should call the batching method if the backend supports it', done => {
+        const metadataGetObjectsStub = sinon.stub(metadataUtils, 'metadataGetObjects').yields(null, {});
+        const objectVersion = 'someVersionId';
+        sinon.stub(multiObjectDelete, 'decodeObjectVersion').returns([null, objectVersion]);
+
+        initializeMultiObjectDeleteWithBatchingSupport(bucketName, inPlay, log, callback);
+
+        assert.strictEqual(metadataGetObjectsStub.callCount, 1);
+        sinon.assert.calledOnce(callback);
+        assert.strictEqual(callback.getCall(0).args[0], null);
+        done();
+    });
+
+    it('should not return an error if the metadataGetObjects function fails', done => {
+        const metadataGetObjectsStub =
+            sinon.stub(metadataUtils, 'metadataGetObjects').yields(new Error('metadata error'), null);
+        const objectVersion = 'someVersionId';
+        sinon.stub(multiObjectDelete, 'decodeObjectVersion').returns([null, objectVersion]);
+
+        initializeMultiObjectDeleteWithBatchingSupport(bucketName, inPlay, log, callback);
+
+        assert.strictEqual(metadataGetObjectsStub.callCount, 1);
+        sinon.assert.calledOnce(callback);
+        assert.strictEqual(callback.getCall(0).args[0] instanceof Error, false);
+        assert.deepStrictEqual(callback.getCall(0).args[1], {});
+        done();
+    });
+
+    it('should populate the cache when the backend supports it', done => {
+        const expectedOutput = {
+            one: {
+                value: 'object1',
+            },
+            two: {
+                value: 'object2',
+            },
+        };
+        const metadataGetObjectsStub = sinon.stub(metadataUtils, 'metadataGetObjects').yields(null, expectedOutput);
+        const objectVersion = 'someVersionId';
+        sinon.stub(multiObjectDelete, 'decodeObjectVersion').returns([null, objectVersion]);
+
+        initializeMultiObjectDeleteWithBatchingSupport(bucketName, inPlay, log, callback);
+
+        assert.strictEqual(metadataGetObjectsStub.callCount, 1);
+        sinon.assert.calledOnce(callback);
+        assert.strictEqual(callback.getCall(0).args[0], null);
+        assert.deepStrictEqual(callback.getCall(0).args[1], expectedOutput);
+        done();
+    });
+});
+
+describe('decodeObjectVersion function helper', () => {
+    it('should throw error for invalid version IDs', () => {
+        const ret = decodeObjectVersion({
+            versionId: '\0',
+        });
+        assert(ret[0].is.NoSuchVersion);
+    });
+
+    it('should return "null" for null versionId', () => {
+        const ret = decodeObjectVersion({
+            versionId: 'null',
+        });
+        assert.strictEqual(ret[0], null);
+        assert.strictEqual(ret[1], 'null');
+    });
+
+    it('should return null error on success', () => {
+        const ret = decodeObjectVersion({});
+        assert.ifError(ret[0]);
+        assert.deepStrictEqual(ret[1], undefined);
+    });
 });
diff --git a/tests/unit/api/objectDelete.js b/tests/unit/api/objectDelete.js
index 39e22739c5..934b03ebb9 100644
--- a/tests/unit/api/objectDelete.js
+++ b/tests/unit/api/objectDelete.js
@@ -142,6 +142,7 @@ describe('objectDelete API', () => {
                             any, any, any,
                             { deleteData: true,
                               replayId: testUploadId,
+                              doesNotNeedOpogUpdate: true,
                             }, any, any, any);
                         done();
                     });
diff --git a/tests/unit/metadata/metadataUtils.spec.js b/tests/unit/metadata/metadataUtils.spec.js
index 574c90e41e..ef83c607bb 100644
--- a/tests/unit/metadata/metadataUtils.spec.js
+++ b/tests/unit/metadata/metadataUtils.spec.js
@@ -1,4 +1,5 @@
 const assert = require('assert');
+const sinon = require('sinon');
 
 const { models } = require('arsenal');
 const { BucketInfo } = models;
@@ -13,7 +14,8 @@ const bucket = new BucketInfo('niftyBucket', ownerCanonicalId,
     authInfo.getAccountDisplayName(), creationDate);
 const log = new DummyRequestLogger();
 
-const { validateBucket } = require('../../../lib/metadata/metadataUtils');
+const { validateBucket, metadataGetObjects, metadataGetObject } = require('../../../lib/metadata/metadataUtils');
+const metadata = require('../../../lib/metadata/wrapper');
 
 describe('validateBucket', () => {
     it('action bucketPutPolicy by bucket owner', () => {
@@ -53,3 +55,94 @@ describe('validateBucket', () => {
         assert(validationResult.is.AccessDenied);
     });
 });
+
+describe('metadataGetObjects', () => {
+    let sandbox;
+    const objectsKeys = [
+        { inPlay: { key: 'objectKey1' }, versionId: 'versionId1' },
+        { inPlay: { key: 'objectKey2' }, versionId: 'versionId2' },
+    ];
+
+    beforeEach(() => {
+        sandbox = sinon.createSandbox();
+    });
+
+    afterEach(() => {
+        sandbox.restore();
+    });
+
+    it('should return error if metadata.getObjectsMD fails', done => {
+        const error = new Error('Failed to get object metadata');
+        sandbox.stub(metadata, 'getObjectsMD').yields(error);
+
+        metadataGetObjects('bucketName', objectsKeys, log, err => {
+            assert(err);
+            assert.strictEqual(err, error);
+            done();
+        });
+    });
+
+    it('should return object metadata if successful', done => {
+        const metadataObjs = [
+            { doc: { key: 'objectKey1' }, versionId: 'versionId1' },
+            { doc: { key: 'objectKey2' }, versionId: 'versionId2' },
+        ];
+        sandbox.stub(metadata, 'getObjectsMD').yields(null, metadataObjs);
+
+        metadataGetObjects('bucketName', objectsKeys, log, (err, result) => {
+            assert.ifError(err);
+            assert(result);
+            assert.strictEqual(result.objectKey1versionId1, metadataObjs[0].doc);
+            assert.strictEqual(result.objectKey2versionId2, metadataObjs[1].doc);
+            done();
+        });
+    });
+});
+
+describe('metadataGetObject', () => {
+    let sandbox;
+    const objectKey = { inPlay: { key: 'objectKey1' }, versionId: 'versionId1' };
+
+    beforeEach(() => {
+        sandbox = sinon.createSandbox();
+    });
+
+    afterEach(() => {
+        sandbox.restore();
+    });
+
+    it('should return the cached document if provided', done => {
+        const cachedDoc = {
+            [objectKey.inPlay.key]: {
+                key: 'objectKey1', versionId: 'versionId1',
+            },
+        };
+        metadataGetObject('bucketName', objectKey.inPlay.key, objectKey.versionId, cachedDoc, log, (err, result) => {
+            assert.ifError(err);
+            assert.deepStrictEqual(result, cachedDoc[objectKey.inPlay.key]);
+            done();
+        });
+    });
+
+    it('should return error if metadata.getObjectMD fails', done => {
+        const error = new Error('Failed to get object metadata');
+        sandbox.stub(metadata, 'getObjectMD').yields(error);
+
+        metadataGetObject('bucketName', objectKey.inPlay.key, objectKey.versionId, null, log, err => {
+            assert(err);
+            assert.strictEqual(err, error);
+            done();
+        });
+    });
+
+    it('should return object metadata if successful', done => {
+        const metadataObj = { doc: { key: 'objectKey1', versionId: 'versionId1' } };
+        sandbox.stub(metadata, 'getObjectMD').yields(null, metadataObj);
+
+        metadataGetObject('bucketName', objectKey.inPlay.key, objectKey.versionId, null, log, (err, result) => {
+            assert.ifError(err);
+            assert.deepStrictEqual(result, metadataObj);
+            done();
+        });
+    });
+});
diff --git a/tests/utilities/objectLock-util.js b/tests/utilities/objectLock-util.js
index e7a1778a4f..92a223e6cd 100644
--- a/tests/utilities/objectLock-util.js
+++ b/tests/utilities/objectLock-util.js
@@ -12,7 +12,7 @@ const log = new DummyRequestLogger();
 function changeObjectLock(objects, newConfig, cb) {
     async.each(objects, (object, next) => {
         const { bucket, key, versionId } = object;
-        metadataGetObject(bucket, key, versionIdUtils.decode(versionId), log, (err, objMD) => {
+        metadataGetObject(bucket, key, versionIdUtils.decode(versionId), null, log, (err, objMD) => {
             assert.ifError(err);
             // set newConfig as empty string to remove object lock
             /* eslint-disable no-param-reassign */
diff --git a/yarn.lock b/yarn.lock
index c2b862f938..008e0d21cc 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -450,9 +450,9 @@
   integrity sha512-pg9d0yC4rVNWQzX8U7xb4olIOFuuVL9za3bzMT2pu2SU0SNEi66i2qrvhE2qt0HvkhuCaWJu7pLNOt/Pj8BIrw==
 
 "@types/node@>=10.0.0":
-  version "20.2.5"
-  resolved "https://registry.yarnpkg.com/@types/node/-/node-20.2.5.tgz#26d295f3570323b2837d322180dfbf1ba156fefb"
-  integrity sha512-JJulVEQXmiY9Px5axXHeYGLSjhkZEnD+MDPDGbCbIAbMslkKwmygtZFy1X6s/075Yo94sf8GuSlFfPzysQrWZQ==
+  version "20.4.2"
+  resolved "https://registry.yarnpkg.com/@types/node/-/node-20.4.2.tgz#129cc9ae69f93824f92fac653eebfb4812ab4af9"
+  integrity sha512-Dd0BYtWgnWJKwO1jkmTrzofjK2QXXcai0dmtzvIBhcA+RsG5h8R3xlyta0kGOZRNfL9GuRtb1knmPEhQrePCEw==
 
 "@types/triple-beam@^1.3.2":
   version "1.3.2"
@@ -789,9 +789,9 @@ arraybuffer.slice@~0.0.7:
   optionalDependencies:
     ioctl "^2.0.2"
 
-"arsenal@git+https://github.com/scality/arsenal#8.1.103":
-  version "8.1.103"
-  resolved "git+https://github.com/scality/arsenal#96cbaeb821d8045cbe8eabd00092290e13e46784"
+"arsenal@git+https://github.com/scality/arsenal#8.1.104":
+  version "8.1.104"
+  resolved "git+https://github.com/scality/arsenal#8716fee67d77f78e79497c75c62d9e4fa5b9a80e"
   dependencies:
     "@azure/identity" "^3.1.1"
     "@azure/storage-blob" "^12.12.0"
@@ -808,7 +808,7 @@ arraybuffer.slice@~0.0.7:
     bson "4.0.0"
     debug "~4.1.0"
     diskusage "^1.1.1"
-    fcntl "github:scality/node-fcntl#0.2.0"
+    fcntl "github:scality/node-fcntl#0.2.2"
     hdclient scality/hdclient#1.1.5
     httpagent scality/httpagent#1.0.6
     https-proxy-agent "^2.2.0"
@@ -2258,6 +2258,14 @@ fastq@^1.6.0:
     nan "^2.3.2"
     node-gyp "^8.0.0"
 
+"fcntl@github:scality/node-fcntl#0.2.2":
+  version "0.2.1"
+  resolved "https://codeload.github.com/scality/node-fcntl/tar.gz/b1335ca204c6265cedc50c26020c4d63aabe920e"
+  dependencies:
+    bindings "^1.1.1"
+    nan "^2.3.2"
+    node-gyp "^8.0.0"
+
 fecha@^4.2.0:
   version "4.2.3"
   resolved "https://registry.yarnpkg.com/fecha/-/fecha-4.2.3.tgz#4d9ccdbc61e8629b259fdca67e65891448d569fd"