diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java
index 721ca77b41b0b..078d9ac27d389 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java
@@ -41,6 +41,7 @@
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -126,6 +127,8 @@ List<Pair<String, HoodieRollbackStat>> maybeDeleteAndCollectStats(HoodieEngineCo
         String latestBaseInstant = rollbackRequest.getLatestBaseInstant();
         FileSystem fs = metaClient.getFs();
         // collect all log files that is supposed to be deleted with this rollback
+        // what happens if file was deleted when invoking fs.getFileStatus(?) below.
+        // I understand we don't delete log files. but just curious if we need to handle this case.
         Map<FileStatus, Long> writtenLogFileSizeMap = new HashMap<>();
         for (Map.Entry<String, Long> entry : logFilesToBeDeleted.entrySet()) {
           writtenLogFileSizeMap.put(fs.getFileStatus(new Path(entry.getKey())), entry.getValue());
@@ -188,7 +191,12 @@ protected List<HoodieRollbackStat> deleteFiles(HoodieTableMetaClient metaClient,
         String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), fullDeletePath.getParent());
         boolean isDeleted = true;
         if (doDelete) {
-          isDeleted = metaClient.getFs().delete(fullDeletePath);
+          try {
+            isDeleted = metaClient.getFs().delete(fullDeletePath);
+          } catch (FileNotFoundException e) {
+            // if first rollback attempt failed and retried again, chances that some files are already deleted.
+            isDeleted = true;
+          }
         }
         return HoodieRollbackStat.newBuilder()
             .withPartitionPath(partitionPath)
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 14fe07b32c1bb..0ee3ba2e04c62 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -195,7 +195,7 @@ public static List<HoodieRecord> convertMetadataToRecords(HoodieCleanMetadata cl
 
     cleanMetadata.getPartitionMetadata().forEach((partition, partitionMetadata) -> {
       // Files deleted from a partition
-      List<String> deletedFiles = partitionMetadata.getSuccessDeleteFiles();
+      List<String> deletedFiles = partitionMetadata.getDeletePathPatterns();
       HoodieRecord record = HoodieMetadataPayload.createPartitionFilesRecord(partition, Option.empty(),
           Option.of(new ArrayList<>(deletedFiles)));
 
@@ -285,7 +285,7 @@ private static void processRollbackMetadata(HoodieActiveTimeline metadataTableTi
       }
 
       final String partition = pm.getPartitionPath();
-      if (!pm.getSuccessDeleteFiles().isEmpty() && !shouldSkip) {
+      if ((!pm.getSuccessDeleteFiles().isEmpty() || !pm.getFailedDeleteFiles().isEmpty()) && !shouldSkip) {
         if (!partitionToDeletedFiles.containsKey(partition)) {
           partitionToDeletedFiles.put(partition, new ArrayList<>());
         }
@@ -293,6 +293,10 @@ private static void processRollbackMetadata(HoodieActiveTimeline metadataTableTi
         // Extract deleted file name from the absolute paths saved in getSuccessDeleteFiles()
         List<String> deletedFiles = pm.getSuccessDeleteFiles().stream().map(p -> new Path(p).getName())
             .collect(Collectors.toList());
+        if (!pm.getFailedDeleteFiles().isEmpty()) {
+          deletedFiles.addAll(pm.getFailedDeleteFiles().stream().map(p -> new Path(p).getName())
+              .collect(Collectors.toList()));
+        }
         partitionToDeletedFiles.get(partition).addAll(deletedFiles);
       }