From 29886f7c80f89a60633ed163ddac6d91a2c27184 Mon Sep 17 00:00:00 2001
From: Ray Mattingly <rmdmattingly@gmail.com>
Date: Mon, 2 Sep 2024 04:38:29 -0400
Subject: [PATCH] HBASE-28697 Don't clean bulk load system entries until backup
 is complete (#6089)

Co-authored-by: Ray Mattingly <rmattingly@hubspot.com>
---
 .../impl/IncrementalTableBackupClient.java      | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalTableBackupClient.java b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalTableBackupClient.java
index 9b9ffcf7a04e..5f6ad97ea547 100644
--- a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalTableBackupClient.java
+++ b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalTableBackupClient.java
@@ -104,13 +104,14 @@ protected static int getIndex(TableName tbl, List<TableName> sTableList) {
 
   /*
    * Reads bulk load records from backup table, iterates through the records and forms the paths for
-   * bulk loaded hfiles. Copies the bulk loaded hfiles to backup destination
+   * bulk loaded hfiles. Copies the bulk loaded hfiles to backup destination. This method does NOT
+   * clean up the entries in the bulk load system table. Those entries should not be cleaned until
+   * the backup is marked as complete.
    * @param sTableList list of tables to be backed up
-   * @return map of table to List of files
+   * @return the rowkeys of bulk loaded files
    */
   @SuppressWarnings("unchecked")
-  protected Map<byte[], List<Path>>[] handleBulkLoad(List<TableName> sTableList)
-    throws IOException {
+  protected List<byte[]> handleBulkLoad(List<TableName> sTableList) throws IOException {
     Map<byte[], List<Path>>[] mapForSrc = new Map[sTableList.size()];
     List<String> activeFiles = new ArrayList<>();
     List<String> archiveFiles = new ArrayList<>();
@@ -192,8 +193,8 @@ protected Map<byte[], List<Path>>[] handleBulkLoad(List<TableName> sTableList)
     }
 
     copyBulkLoadedFiles(activeFiles, archiveFiles);
-    backupManager.deleteBulkLoadedRows(pair.getSecond());
-    return mapForSrc;
+
+    return pair.getSecond();
   }
 
   private void copyBulkLoadedFiles(List<String> activeFiles, List<String> archiveFiles)
@@ -309,10 +310,12 @@ public void execute() throws IOException {
         BackupUtils.getMinValue(BackupUtils.getRSLogTimestampMins(newTableSetTimestampMap));
       backupManager.writeBackupStartCode(newStartCode);
 
-      handleBulkLoad(backupInfo.getTableNames());
+      List<byte[]> bulkLoadedRows = handleBulkLoad(backupInfo.getTableNames());
+
       // backup complete
       completeBackup(conn, backupInfo, BackupType.INCREMENTAL, conf);
 
+      backupManager.deleteBulkLoadedRows(bulkLoadedRows);
     } catch (IOException e) {
       failBackup(conn, backupInfo, backupManager, e, "Unexpected Exception : ",
         BackupType.INCREMENTAL, conf);