diff --git a/ambry-api/src/main/java/com/github/ambry/store/StoreErrorCodes.java b/ambry-api/src/main/java/com/github/ambry/store/StoreErrorCodes.java
index 4464998bc7..af13d31543 100644
--- a/ambry-api/src/main/java/com/github/ambry/store/StoreErrorCodes.java
+++ b/ambry-api/src/main/java/com/github/ambry/store/StoreErrorCodes.java
@@ -39,5 +39,7 @@ public enum StoreErrorCodes {
   Life_Version_Conflict,
   ID_Not_Deleted,
   ID_Undeleted,
-  ID_Deleted_Permanently
+  ID_Deleted_Permanently,
+  // TODO Efficient_Metadata_Operations_TODO : ID_Purged error should be handled in all of the store's methods.
+  ID_Purged
 }
diff --git a/ambry-commons/src/main/java/com/github/ambry/commons/ServerMetrics.java b/ambry-commons/src/main/java/com/github/ambry/commons/ServerMetrics.java
index e7c7cfe843..ac95f1030d 100644
--- a/ambry-commons/src/main/java/com/github/ambry/commons/ServerMetrics.java
+++ b/ambry-commons/src/main/java/com/github/ambry/commons/ServerMetrics.java
@@ -267,6 +267,7 @@ public class ServerMetrics {
   public final Counter unknownFormatError;
   public final Counter idNotFoundError;
   public final Counter idDeletedError;
+  public final Counter idPurgedError;
   public final Counter idUndeletedError;
   public final Counter idNotDeletedError;
   public final Counter lifeVersionConflictError;
@@ -622,6 +623,7 @@ public ServerMetrics(MetricRegistry registry, Class<?> requestClass, Class<?> se
     unknownFormatError = registry.counter(MetricRegistry.name(requestClass, "UnknownFormatError"));
     idNotFoundError = registry.counter(MetricRegistry.name(requestClass, "IDNotFoundError"));
     idDeletedError = registry.counter(MetricRegistry.name(requestClass, "IDDeletedError"));
+    idPurgedError = registry.counter(MetricRegistry.name(requestClass, "IDPurgedError"));
     idUndeletedError = registry.counter(MetricRegistry.name(requestClass, "IDUndeletedError"));
     idNotDeletedError = registry.counter(MetricRegistry.name(requestClass, "IDNotDeletedError"));
     lifeVersionConflictError = registry.counter(MetricRegistry.name(requestClass, "lifeVersionConflictError"));
diff --git a/ambry-server/src/integration-test/java/com/github/ambry/server/MockCluster.java b/ambry-server/src/integration-test/java/com/github/ambry/server/MockCluster.java
index a731a2cbe7..c56fe8a3f3 100644
--- a/ambry-server/src/integration-test/java/com/github/ambry/server/MockCluster.java
+++ b/ambry-server/src/integration-test/java/com/github/ambry/server/MockCluster.java
@@ -523,7 +523,7 @@ void trackDeletion(String host, int port) {
 
   /**
    * Tracks the purge event that arrived on {@code host}:{@code port}.
-   * @param host the host that received the compact message.
+   * @param host the host that received the purge request.
    * @param port the port of the host that describes the instance along with {@code host}.
    */
   void trackPurge(String host, int port) {
diff --git a/ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java b/ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java
index 9bc0431b8a..8e76d858f2 100644
--- a/ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java
+++ b/ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java
@@ -60,6 +60,8 @@
 import com.github.ambry.protocol.GetResponse;
 import com.github.ambry.protocol.PartitionRequestInfo;
 import com.github.ambry.protocol.PartitionResponseInfo;
+import com.github.ambry.protocol.PurgeRequest;
+import com.github.ambry.protocol.PurgeResponse;
 import com.github.ambry.protocol.PutRequest;
 import com.github.ambry.protocol.PutResponse;
 import com.github.ambry.protocol.ReplicaMetadataRequest;
@@ -506,7 +508,7 @@ public void handlePurgeRequest(NetworkRequest request) throws IOException, Inter
       ServerErrorCode error =
           validateRequest(purgeRequest.getBlobId().getPartition(), RequestOrResponseType.PurgeRequest, false);
       if (error != ServerErrorCode.No_Error) {
-        logger.error("Validating delete request failed with error {} for request {}", error, purgeRequest);
+        logger.error("Validating purge request failed with error {} for request {}", error, purgeRequest);
         response = new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(), error);
       } else {
         BlobId convertedBlobId = (BlobId) convertedStoreKey;
@@ -514,8 +516,8 @@ public void handlePurgeRequest(NetworkRequest request) throws IOException, Inter
             convertedBlobId.getContainerId(), purgeRequest.getPurgeTimeInMs()).isDeleted(true)
             .lifeVersion(MessageInfo.LIFE_VERSION_FROM_FRONTEND)
             .build();
-        Store storeToDelete = storeManager.getStore(purgeRequest.getBlobId().getPartition());
-        storeToDelete.purge(Collections.singletonList(info));
+        Store storeToPurge = storeManager.getStore(purgeRequest.getBlobId().getPartition());
+        storeToPurge.purge(Collections.singletonList(info));
         response =
             new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(), ServerErrorCode.No_Error);
         if (notification != null) {
@@ -527,27 +529,25 @@ public void handlePurgeRequest(NetworkRequest request) throws IOException, Inter
       boolean logInErrorLevel = false;
       if (e.getErrorCode() == StoreErrorCodes.ID_Not_Found) {
         metrics.idNotFoundError.inc();
-      } else if (e.getErrorCode() == StoreErrorCodes.TTL_Expired) {
-        metrics.ttlExpiredError.inc();
-      } else if (e.getErrorCode() == StoreErrorCodes.ID_Deleted) {
-        metrics.idDeletedError.inc();
+      } else if (e.getErrorCode() == StoreErrorCodes.ID_Purged) {
+        metrics.idPurgedError.inc();
       } else if (e.getErrorCode() == StoreErrorCodes.Authorization_Failure) {
         metrics.purgeAuthorizationFailure.inc();
       } else {
         logInErrorLevel = true;
-        metrics.unExpectedStoreDeleteError.inc();
+        metrics.unExpectedStorePurgeError.inc();
       }
       if (logInErrorLevel) {
-        logger.error("Store exception on a compact message with error code {} for request {}", e.getErrorCode(), purgeRequest,
-            e);
+        logger.error("Store exception on a purge message with error code {} for request {}", e.getErrorCode(),
+            purgeRequest, e);
       } else {
-        logger.trace("Store exception on a compact with error code {} for request {}", e.getErrorCode(), purgeRequest,
+        logger.trace("Store exception on a purge with error code {} for request {}", e.getErrorCode(), purgeRequest,
             e);
       }
       response = new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(),
           ErrorMapping.getStoreErrorMapping(e.getErrorCode()));
     } catch (Exception e) {
-      logger.error("Unknown exception for compact message request {}", purgeRequest, e);
+      logger.error("Unknown exception for purge request {}", purgeRequest, e);
       response = new PurgeResponse(purgeRequest.getCorrelationId(), purgeRequest.getClientId(),
           ServerErrorCode.Unknown_Error);
       metrics.unExpectedStorePurgeError.inc();