diff --git a/core/src/main/java/com/datastrato/gravitino/storage/kv/KvGarbageCollector.java b/core/src/main/java/com/datastrato/gravitino/storage/kv/KvGarbageCollector.java
index 7984abdb83c..309323f490c 100644
--- a/core/src/main/java/com/datastrato/gravitino/storage/kv/KvGarbageCollector.java
+++ b/core/src/main/java/com/datastrato/gravitino/storage/kv/KvGarbageCollector.java
@@ -193,14 +193,15 @@ private void collectAndRemoveOldVersionData() throws IOException {
           removeAllVersionsOfKey(rawKey, key, false);
 
           LogHelper logHelper = decodeKey(key, transactionId);
+          kvBackend.delete(rawKey);
           LOG.info(
-              "Physically delete key that has marked deleted: name identifier: '{}', entity type: '{}', createTime: '{}({})', key: '{}'",
+              "Physically delete key that has marked deleted: name identifier: '{}', entity type: '{}',"
+                  + " createTime: '{}({})', key: '{}'",
               logHelper.identifier,
               logHelper.type,
               logHelper.createTimeAsString,
               logHelper.createTimeInMs,
               Bytes.wrap(key));
-          kvBackend.delete(rawKey);
           keysDeletedCount++;
           continue;
         }
@@ -225,8 +226,10 @@ private void collectAndRemoveOldVersionData() throws IOException {
           LogHelper logHelper = decodeKey(key, transactionId);
           byte[] newVersionKey = newVersionOfKey.get(0).getKey();
           LogHelper newVersionLogHelper = decodeKey(newVersionKey);
+          kvBackend.delete(rawKey);
           LOG.info(
-              "Physically delete key that has newer version: name identifier: '{}', entity type: '{}', createTime: '{}({})', newVersion createTime: '{}({})',"
+              "Physically delete key that has newer version: name identifier: '{}', entity type: '{}',"
+                  + " createTime: '{}({})', newVersion createTime: '{}({})',"
                   + " key: '{}', newVersion key: '{}'",
               logHelper.identifier,
               logHelper.type,
@@ -236,13 +239,13 @@ private void collectAndRemoveOldVersionData() throws IOException {
               newVersionLogHelper.createTimeInMs,
               Bytes.wrap(rawKey),
               Bytes.wrap(newVersionKey));
-          kvBackend.delete(rawKey);
           keysDeletedCount++;
         }
       }
 
       // All keys in this transaction have been deleted, we can remove the commit mark.
       if (keysDeletedCount == keysInTheTransaction.size()) {
+        kvBackend.delete(kv.getKey());
         long timestamp = getTransactionId(transactionId) >> 18;
         LOG.info(
             "Physically delete commit mark: {}, createTime: '{}({})', key: '{}'",
@@ -250,7 +253,6 @@ private void collectAndRemoveOldVersionData() throws IOException {
             DateFormatUtils.format(timestamp, TIME_STAMP_FORMAT),
             timestamp,
             Bytes.wrap(kv.getKey()));
-        kvBackend.delete(kv.getKey());
       }
     }
 
@@ -283,7 +285,8 @@ private void removeAllVersionsOfKey(byte[] rawKey, byte[] key, boolean includeSt
 
       LogHelper logHelper = decodeKey(kv.getKey());
       LOG.info(
-          "Physically delete key that has marked deleted: name identifier: '{}', entity type: '{}', createTime: '{}({})', key: '{}'",
+          "Physically delete key that has marked deleted: name identifier: '{}', entity type: '{}',"
+              + " createTime: '{}({})', key: '{}'",
           logHelper.identifier,
           logHelper.type,
           logHelper.createTimeAsString,
@@ -308,6 +311,7 @@ private void removeAllVersionsOfKey(byte[] rawKey, byte[] key, boolean includeSt
 
       // Try to delete the commit mark.
       if (allDropped) {
+        kvBackend.delete(transactionKey);
         long timestamp = TransactionalKvBackendImpl.getTransactionId(transactionId) >> 18;
         LOG.info(
             "Physically delete commit mark: {}, createTime: '{}({})', key: '{}'",
@@ -315,7 +319,6 @@ private void removeAllVersionsOfKey(byte[] rawKey, byte[] key, boolean includeSt
             DateFormatUtils.format(timestamp, TIME_STAMP_FORMAT),
             timestamp,
             Bytes.wrap(kv.getKey()));
-        kvBackend.delete(transactionKey);
       }
     }
   }