diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocol/KeySpaceManagerProtocol.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocol/KeySpaceManagerProtocol.java
index 6c743e27df6..10b69753b30 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocol/KeySpaceManagerProtocol.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocol/KeySpaceManagerProtocol.java
@@ -146,4 +146,12 @@ public interface KeySpaceManagerProtocol {
    */
   KsmKeyInfo lookupKey(KsmKeyArgs args) throws IOException;
 
+  /**
+   * Deletes an existing key.
+   *
+   * @param args the args of the key.
+   * @throws IOException
+   */
+  void deleteKey(KsmKeyArgs args) throws IOException;
+
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocolPB/KeySpaceManagerProtocolClientSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocolPB/KeySpaceManagerProtocolClientSideTranslatorPB.java
index bf33b2b2e2c..26a3bb9f9fb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocolPB/KeySpaceManagerProtocolClientSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/ksm/protocolPB/KeySpaceManagerProtocolClientSideTranslatorPB.java
@@ -445,6 +445,33 @@ public final class KeySpaceManagerProtocolClientSideTranslatorPB
     return KsmKeyInfo.getFromProtobuf(resp.getKeyInfo());
   }
 
+  /**
+   * Deletes an existing key.
+   *
+   * @param args the args of the key.
+   * @throws IOException
+   */
+  @Override
+  public void deleteKey(KsmKeyArgs args) throws IOException {
+    LocateKeyRequest.Builder req = LocateKeyRequest.newBuilder();
+    KeyArgs keyArgs = KeyArgs.newBuilder()
+        .setVolumeName(args.getVolumeName())
+        .setBucketName(args.getBucketName())
+        .setKeyName(args.getKeyName()).build();
+    req.setKeyArgs(keyArgs);
+
+    final LocateKeyResponse resp;
+    try {
+      resp = rpcProxy.deleteKey(NULL_RPC_CONTROLLER, req.build());
+    } catch (ServiceException e) {
+      throw ProtobufHelper.getRemoteException(e);
+    }
+    if (resp.getStatus() != Status.OK) {
+      throw new IOException("Delete key failed, error:" +
+          resp.getStatus());
+    }
+  }
+
   /**
    * Return the proxy object underlying this protocol translator.
    *
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/protocolPB/ScmBlockLocationProtocolClientSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/protocolPB/ScmBlockLocationProtocolClientSideTranslatorPB.java
index fa6529948f2..4a0d50be6cc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/protocolPB/ScmBlockLocationProtocolClientSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/protocolPB/ScmBlockLocationProtocolClientSideTranslatorPB.java
@@ -155,7 +155,7 @@ public final class ScmBlockLocationProtocolClientSideTranslatorPB
   @Override
   public List<DeleteBlockResult> deleteBlocks(Set<String> keys)
       throws IOException {
-    Preconditions.checkArgument(keys == null || keys.isEmpty(),
+    Preconditions.checkArgument(keys != null && !keys.isEmpty(),
         "keys to be deleted cannot be null or empty");
     DeleteScmBlocksRequestProto request = DeleteScmBlocksRequestProto
         .newBuilder()
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/KeySpaceManagerProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/KeySpaceManagerProtocol.proto
index 0a6f7bc7485..130f59a2388 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/KeySpaceManagerProtocol.proto
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/KeySpaceManagerProtocol.proto
@@ -218,7 +218,7 @@ message KeyArgs {
     required string volumeName = 1;
     required string bucketName = 2;
     required string keyName = 3;
-    required uint64 dataSize = 4;
+    optional uint64 dataSize = 4;
 }
 
 message KeyInfo {
@@ -317,4 +317,10 @@ service KeySpaceManagerService {
     */
     rpc lookupKey(LocateKeyRequest)
         returns(LocateKeyResponse);
+
+    /**
+       Delete an existing key.
+    */
+    rpc deleteKey(LocateKeyRequest)
+        returns(LocateKeyResponse);
 }
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KSMMetrics.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KSMMetrics.java
index 27a14cee9ed..becaf4ece6b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KSMMetrics.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KSMMetrics.java
@@ -38,6 +38,7 @@ public class KSMMetrics {
   private @Metric MutableCounterLong numBucketModifies;
   private @Metric MutableCounterLong numKeyAllocate;
   private @Metric MutableCounterLong numKeyLookup;
+  private @Metric MutableCounterLong numKeyDeletes;
 
   // Failure Metrics
   private @Metric MutableCounterLong numVolumeCreateFails;
@@ -50,6 +51,7 @@ public class KSMMetrics {
   private @Metric MutableCounterLong numBucketModifyFails;
   private @Metric MutableCounterLong numKeyAllocateFails;
   private @Metric MutableCounterLong numKeyLookupFails;
+  private @Metric MutableCounterLong numKeyDeleteFails;
 
   public KSMMetrics() {
   }
@@ -141,6 +143,14 @@ public class KSMMetrics {
     numKeyLookupFails.incr();
   }
 
+  public void incNumKeyDeleteFails() {
+    numKeyDeleteFails.incr();
+  }
+
+  public void incNumKeyDeletes() {
+    numKeyDeletes.incr();
+  }
+
   @VisibleForTesting
   public long getNumVolumeCreates() {
     return numVolumeCreates.value();
@@ -240,4 +250,14 @@ public class KSMMetrics {
   public long getNumKeyLookupFails() {
     return numKeyLookupFails.value();
   }
+
+  @VisibleForTesting
+  public long getNumKeyDeletes() {
+    return numKeyDeletes.value();
+  }
+
+  @VisibleForTesting
+  public long getNumKeyDeletesFails() {
+    return numKeyDeleteFails.value();
+  }
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManager.java
index d11069e483e..8ea9df32a47 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManager.java
@@ -52,4 +52,15 @@ public interface KeyManager {
    * @throws IOException
    */
   KsmKeyInfo lookupKey(KsmKeyArgs args) throws IOException;
+
+  /**
+   * Deletes an object by an object key. The key will be immediately removed
+   * from KSM namespace and become invisible to clients. The object data
+   * will be removed in async manner that might retain for some time.
+   *
+   * @param args the args of the key provided by client.
+   * @throws IOException if specified key doesn't exist or
+   * some other I/O errors while deleting an object.
+   */
+  void deleteKey(KsmKeyArgs args) throws IOException;
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java
index 555ca9d801b..deba4a39c55 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeyManagerImpl.java
@@ -20,14 +20,19 @@ import com.google.common.base.Preconditions;
 import org.apache.hadoop.ksm.helpers.KsmKeyArgs;
 import org.apache.hadoop.ksm.helpers.KsmKeyInfo;
 import org.apache.hadoop.ozone.ksm.exceptions.KSMException;
+import org.apache.hadoop.ozone.ksm.exceptions.KSMException.ResultCodes;
 import org.apache.hadoop.ozone.protocol.proto.KeySpaceManagerProtocolProtos.KeyInfo;
+import org.apache.hadoop.ozone.protocol.proto.ScmBlockLocationProtocolProtos.DeleteScmBlockResult.Result;
 import org.apache.hadoop.scm.container.common.helpers.AllocatedBlock;
+import org.apache.hadoop.scm.container.common.helpers.DeleteBlockResult;
 import org.apache.hadoop.scm.protocol.ScmBlockLocationProtocol;
 import org.iq80.leveldb.DBException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
 
 /**
  * Implementation of keyManager.
@@ -139,4 +144,40 @@ public class KeyManagerImpl implements KeyManager {
       metadataManager.writeLock().unlock();
     }
   }
+
+  @Override
+  public void deleteKey(KsmKeyArgs args) throws IOException {
+    Preconditions.checkNotNull(args);
+    KsmKeyInfo keyInfo = lookupKey(args);
+
+    metadataManager.writeLock().lock();
+    String volumeName = args.getVolumeName();
+    String bucketName = args.getBucketName();
+    String keyName = args.getKeyName();
+    try {
+      List<DeleteBlockResult> resultList =
+          scmBlockClient.deleteBlocks(
+              Collections.singleton(keyInfo.getBlockID()));
+      if (resultList.size() != 1) {
+        throw new KSMException("Delete result size from SCM is wrong",
+            ResultCodes.FAILED_INTERNAL_ERROR);
+      }
+
+      if (resultList.get(0).getResult() == Result.success) {
+        byte[] objectKey = metadataManager.getDBKeyForKey(
+            volumeName, bucketName, keyName);
+        metadataManager.deleteKey(objectKey);
+      } else {
+        throw new KSMException("Cannot delete key from SCM",
+                ResultCodes.FAILED_INTERNAL_ERROR);
+      }
+    } catch (DBException ex) {
+      LOG.error(String.format("Delete key failed for volume:%s "
+          + "bucket:%s key:%s", volumeName, bucketName, keyName), ex);
+      throw new KSMException(ex.getMessage(), ex,
+          ResultCodes.FAILED_INTERNAL_ERROR);
+    } finally {
+      metadataManager.writeLock().unlock();
+    }
+  }
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java
index fb991bd07c4..3cf1fb39b4f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/KeySpaceManager.java
@@ -460,6 +460,23 @@ public class KeySpaceManager implements KeySpaceManagerProtocol {
     }
   }
 
+  /**
+   * Deletes an existing key.
+   *
+   * @param args - attributes of the key.
+   * @throws IOException
+   */
+  @Override
+  public void deleteKey(KsmKeyArgs args) throws IOException {
+    try {
+      metrics.incNumKeyDeletes();
+      keyManager.deleteKey(args);
+    } catch (Exception ex) {
+      metrics.incNumKeyDeleteFails();
+      throw ex;
+    }
+  }
+
   /**
    * Sets bucket property from args.
    * @param args - BucketArgs.
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManager.java
index cba2dc953ca..662e0b2d6a6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManager.java
@@ -105,6 +105,13 @@ public interface MetadataManager {
    */
   byte[] getDBKeyForKey(String volume, String bucket, String key);
 
+  /**
+   * Deletes the key from DB.
+   *
+   * @param key - key name
+   */
+  void deleteKey(byte[] key);
+
   /**
    * Given a volume, check if it is empty, i.e there are no buckets inside it.
    * @param volume - Volume name
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java
index 043175efbd0..9e1f2164bd2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/ksm/MetadataManagerImpl.java
@@ -114,6 +114,16 @@ public class MetadataManagerImpl implements  MetadataManager {
     return DFSUtil.string2Bytes(keyKeyString);
   }
 
+  /**
+   * Deletes the key on Metadata DB.
+   *
+   * @param key - key name
+   */
+  @Override
+  public void deleteKey(byte[] key) {
+    store.delete(key);
+  }
+
   /**
    * Returns the read lock used on Metadata DB.
    * @return readLock
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/KeySpaceManagerProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/KeySpaceManagerProtocolServerSideTranslatorPB.java
index 947378ce154..65f3bdf2c3b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/KeySpaceManagerProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/protocolPB/KeySpaceManagerProtocolServerSideTranslatorPB.java
@@ -313,4 +313,24 @@ public class KeySpaceManagerProtocolServerSideTranslatorPB implements
     }
     return resp.build();
   }
+
+  @Override
+  public LocateKeyResponse deleteKey(RpcController controller,
+      LocateKeyRequest request) throws ServiceException {
+    LocateKeyResponse.Builder resp =
+        LocateKeyResponse.newBuilder();
+    try {
+      KeyArgs keyArgs = request.getKeyArgs();
+      KsmKeyArgs ksmKeyArgs = new KsmKeyArgs.Builder()
+          .setVolumeName(keyArgs.getVolumeName())
+          .setBucketName(keyArgs.getBucketName())
+          .setKeyName(keyArgs.getKeyName())
+          .build();
+      impl.deleteKey(ksmKeyArgs);
+      resp.setStatus(Status.OK);
+    } catch (IOException e) {
+      resp.setStatus(exceptionToResponseStatus(e));
+    }
+    return resp.build();
+  }
 }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/handlers/KeyArgs.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/handlers/KeyArgs.java
index 754c3339782..48a4cb45136 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/handlers/KeyArgs.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/handlers/KeyArgs.java
@@ -23,7 +23,6 @@ package org.apache.hadoop.ozone.web.handlers;
  */
 public class KeyArgs extends BucketArgs {
   private String key;
-  private boolean delete;
   private String hash;
   private long size;
 
@@ -40,6 +39,17 @@ public class KeyArgs extends BucketArgs {
     this.key = objectName;
   }
 
+  /**
+   * Constructor for Key Args.
+   *
+   * @param objectName - Key
+   * @param args - Bucket Args
+   */
+  public KeyArgs(String objectName, BucketArgs args) {
+    super(args);
+    this.key = objectName;
+  }
+
   /**
    * Get Key Name.
    *
@@ -49,24 +59,6 @@ public class KeyArgs extends BucketArgs {
     return this.key;
   }
 
-  /**
-   * Checks if this request is for a Delete key.
-   *
-   * @return boolean
-   */
-  public boolean isDelete() {
-    return delete;
-  }
-
-  /**
-   * Sets the key request as a Delete Request.
-   *
-   * @param delete bool, indicating if this is a delete request
-   */
-  public void setDelete(boolean delete) {
-    this.delete = delete;
-  }
-
   /**
    * Computed File hash.
    *
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java
index b52e22e38f1..62e64b4ca35 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java
@@ -398,7 +398,12 @@ public final class DistributedStorageHandler implements StorageHandler {
 
   @Override
   public void deleteKey(KeyArgs args) throws IOException, OzoneException {
-    throw new UnsupportedOperationException("deleteKey not implemented");
+    KsmKeyArgs keyArgs = new KsmKeyArgs.Builder()
+        .setVolumeName(args.getVolumeName())
+        .setBucketName(args.getBucketName())
+        .setKeyName(args.getKeyName())
+        .build();
+    keySpaceManagerClient.deleteKey(keyArgs);
   }
 
   @Override
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java
index b9cf3554ef9..a460b87f112 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/ksm/TestKeySpaceManager.java
@@ -47,9 +47,10 @@ import org.junit.rules.ExpectedException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.Random;
-import java.util.List;
+import java.util.Set;
 
 /**
  * Test Key Space Manager operation in distributed handler scenario.
@@ -450,4 +451,54 @@ public class TestKeySpaceManager {
     Assert.assertEquals(1 + numKeyLookupFails,
         ksmMetrics.getNumKeyLookupFails());
   }
+
+  /**
+   * Test delete keys for ksm.
+   *
+   * @throws IOException
+   * @throws OzoneException
+   */
+  @Test
+  public void testDeleteKey() throws IOException, OzoneException {
+    String userName = "user" + RandomStringUtils.randomNumeric(5);
+    String adminName = "admin" + RandomStringUtils.randomNumeric(5);
+    String volumeName = "volume" + RandomStringUtils.randomNumeric(5);
+    String bucketName = "bucket" + RandomStringUtils.randomNumeric(5);
+    String keyName = "key" + RandomStringUtils.randomNumeric(5);
+    long numKeyDeletes = ksmMetrics.getNumKeyDeletes();
+    long numKeyDeleteFails = ksmMetrics.getNumKeyDeletesFails();
+
+    VolumeArgs createVolumeArgs = new VolumeArgs(volumeName, userArgs);
+    createVolumeArgs.setUserName(userName);
+    createVolumeArgs.setAdminName(adminName);
+    storageHandler.createVolume(createVolumeArgs);
+
+    BucketArgs bucketArgs = new BucketArgs(bucketName, createVolumeArgs);
+    storageHandler.createBucket(bucketArgs);
+
+    KeyArgs keyArgs = new KeyArgs(keyName, bucketArgs);
+    keyArgs.setSize(100);
+    String dataString = RandomStringUtils.randomAscii(100);
+    try (OutputStream stream = storageHandler.newKeyWriter(keyArgs)) {
+      stream.write(dataString.getBytes());
+    }
+
+    storageHandler.deleteKey(keyArgs);
+    Assert.assertEquals(1 + numKeyDeletes, ksmMetrics.getNumKeyDeletes());
+
+    // Check the block key in SCM, make sure it's deleted.
+    Set<String> keys = new HashSet<>();
+    keys.add(keyArgs.getResourceName());
+    exception.expect(IOException.class);
+    exception.expectMessage("Specified block key does not exist");
+    cluster.getStorageContainerManager().getBlockLocations(keys);
+
+    // Delete the key again to test deleting non-existing key.
+    exception.expect(IOException.class);
+    exception.expectMessage("KEY_NOT_FOUND");
+    storageHandler.deleteKey(keyArgs);
+    Assert.assertEquals(1 + numKeyDeleteFails,
+        ksmMetrics.getNumKeyDeletesFails());
+  }
+
 }