HBASE-29891: Multi-table continuous incremental backup is failing bec…

hbase-backup/pom.xml

@@ -182,6 +182,11 @@
       <artifactId>mockito-inline</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-inline</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/PointInTimeRestoreDriver.java

@@ -27,7 +27,6 @@
 
 import java.net.URI;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.TableName;
@@ -102,15 +101,9 @@ protected int executeRestore(boolean check, TableName[] fromTables, TableName[]
         return -5;
       }
 
-      // TODO: Currently hardcoding keepOriginalSplits=false and restoreRootDir via tmp dir.
-      // These should come from user input (same issue exists in normal restore).
-      // Expose them as configurable options in future.
-      PointInTimeRestoreRequest pointInTimeRestoreRequest =
-        new PointInTimeRestoreRequest.Builder().withBackupRootDir(backupRootDir).withCheck(check)
-          .withFromTables(fromTables).withToTables(toTables).withOverwrite(isOverwrite)
-          .withToDateTime(endTime).withKeepOriginalSplits(false).withRestoreRootDir(
-            BackupUtils.getTmpRestoreOutputDir(FileSystem.get(conf), conf).toString())
-          .build();
+      PointInTimeRestoreRequest pointInTimeRestoreRequest = new PointInTimeRestoreRequest.Builder()
+        .withBackupRootDir(backupRootDir).withCheck(check).withFromTables(fromTables)
+        .withToTables(toTables).withOverwrite(isOverwrite).withToDateTime(endTime).build();
 
       client.pointInTimeRestore(pointInTimeRestoreRequest);
     } catch (Exception e) {

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/PointInTimeRestoreRequest.java

@@ -27,33 +27,25 @@
 public final class PointInTimeRestoreRequest {
 
   private final String backupRootDir;
-  private final String restoreRootDir;
   private final boolean check;
   private final TableName[] fromTables;
   private final TableName[] toTables;
   private final boolean overwrite;
   private final long toDateTime;
-  private final boolean isKeepOriginalSplits;
 
   private PointInTimeRestoreRequest(Builder builder) {
     this.backupRootDir = builder.backupRootDir;
-    this.restoreRootDir = builder.restoreRootDir;
     this.check = builder.check;
     this.fromTables = builder.fromTables;
     this.toTables = builder.toTables;
     this.overwrite = builder.overwrite;
     this.toDateTime = builder.toDateTime;
-    this.isKeepOriginalSplits = builder.isKeepOriginalSplits;
   }
 
   public String getBackupRootDir() {
     return backupRootDir;
   }
 
-  public String getRestoreRootDir() {
-    return restoreRootDir;
-  }
-
   public boolean isCheck() {
     return check;
   }
@@ -74,30 +66,19 @@ public long getToDateTime() {
     return toDateTime;
   }
 
-  public boolean isKeepOriginalSplits() {
-    return isKeepOriginalSplits;
-  }
-
   public static class Builder {
     private String backupRootDir;
-    private String restoreRootDir;
     private boolean check = false;
     private TableName[] fromTables;
     private TableName[] toTables;
     private boolean overwrite = false;
     private long toDateTime;
-    private boolean isKeepOriginalSplits;
 
     public Builder withBackupRootDir(String backupRootDir) {
       this.backupRootDir = backupRootDir;
       return this;
     }
 
-    public Builder withRestoreRootDir(String restoreRootDir) {
-      this.restoreRootDir = restoreRootDir;
-      return this;
-    }
-
     public Builder withCheck(boolean check) {
       this.check = check;
       return this;
@@ -123,11 +104,6 @@ public Builder withToDateTime(long dateTime) {
       return this;
     }
 
-    public Builder withKeepOriginalSplits(boolean isKeepOriginalSplits) {
-      this.isKeepOriginalSplits = isKeepOriginalSplits;
-      return this;
-    }
-
     public PointInTimeRestoreRequest build() {
       return new PointInTimeRestoreRequest(this);
     }

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/AbstractPitrRestoreHandler.java

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_PITR_WINDOW_DAYS;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_WAL_DIR;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_CONTINUOUS_BACKUP_PITR_WINDOW_DAYS;
+import static org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2.MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY;
 import static org.apache.hadoop.hbase.mapreduce.WALPlayer.IGNORE_EMPTY_FILES;
 
 import java.io.IOException;
@@ -33,9 +34,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.backup.BackupRestoreFactory;
 import org.apache.hadoop.hbase.backup.PointInTimeRestoreRequest;
-import org.apache.hadoop.hbase.backup.RestoreJob;
 import org.apache.hadoop.hbase.backup.RestoreRequest;
 import org.apache.hadoop.hbase.backup.util.BackupUtils;
 import org.apache.hadoop.hbase.client.Connection;
@@ -411,6 +410,9 @@ private Tool initializeWalPlayer(long startTime, long endTime) {
     conf.setLong(WALInputFormat.START_TIME_KEY, startTime);
     conf.setLong(WALInputFormat.END_TIME_KEY, endTime);
     conf.setBoolean(IGNORE_EMPTY_FILES, true);
+    // HFile output format defaults to false in HFileOutputFormat2, but we are explicitly setting
+    // it here just in case
+    conf.setBoolean(MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY, false);
     Tool walPlayer = new WALPlayer();
     walPlayer.setConf(conf);
     return walPlayer;

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupAdminImpl.java

@@ -17,11 +17,6 @@
  */
 package org.apache.hadoop.hbase.backup.impl;
 
-import static org.apache.hadoop.hbase.backup.BackupInfo.withRoot;
-import static org.apache.hadoop.hbase.backup.BackupInfo.withState;
-import static org.apache.hadoop.hbase.backup.BackupInfo.withType;
-import static org.apache.hadoop.hbase.backup.impl.BackupSystemTable.Order.NEW_TO_OLD;
-
 import com.google.errorprone.annotations.RestrictedApi;
 import java.io.IOException;
 import java.util.ArrayList;

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupCommands.java

@@ -17,7 +17,6 @@
  */
 package org.apache.hadoop.hbase.backup.impl;
 
-import static org.apache.hadoop.hbase.backup.BackupInfo.withState;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_PITR_WINDOW_DAYS;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_WAL_DIR;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONTINUOUS_BACKUP_REPLICATION_PEER;
@@ -50,7 +49,6 @@
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_WORKERS_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_YARN_QUEUE_NAME;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_YARN_QUEUE_NAME_DESC;
-import static org.apache.hadoop.hbase.backup.impl.BackupSystemTable.Order.NEW_TO_OLD;
 import static org.apache.hadoop.hbase.backup.replication.ContinuousBackupReplicationEndpoint.ONE_DAY_IN_MILLISECONDS;
 import static org.apache.hadoop.hbase.backup.util.BackupUtils.DATE_FORMAT;
 
@@ -83,7 +81,7 @@
 import org.apache.hadoop.hbase.backup.BackupRestoreConstants.BackupCommand;
 import org.apache.hadoop.hbase.backup.BackupType;
 import org.apache.hadoop.hbase.backup.HBackupFileSystem;
-import org.apache.hadoop.hbase.backup.util.BackupFileSystemManager;
+import org.apache.hadoop.hbase.backup.replication.BackupFileSystemManager;
 import org.apache.hadoop.hbase.backup.util.BackupSet;
 import org.apache.hadoop.hbase.backup.util.BackupUtils;
 import org.apache.hadoop.hbase.client.Admin;
@@ -803,8 +801,7 @@ private List<TableName> getTablesDependentOnBackupForPITR(String backupId) throw
           }
 
           // Check if there is any other valid backup that can cover the PITR window
-          List<BackupInfo> allBackups =
-            backupSystemTable.getBackupHistory(withState(BackupInfo.BackupState.COMPLETE));
+          List<BackupInfo> allBackups = backupSystemTable.getBackupInfos(BackupState.COMPLETE);
           boolean hasAnotherValidBackup =
             canAnyOtherBackupCover(allBackups, targetBackup, table, coveredPitrWindow.get(),
               continuousBackupStartTimes.get(table), maxAllowedPITRTime, currentTime);
@@ -896,8 +893,7 @@ private boolean canAnyOtherBackupCover(List<BackupInfo> allBackups, BackupInfo c
 
     /**
      * Cleans up Write-Ahead Logs (WALs) that are no longer required for PITR after a successful
-     * backup deletion. If no full backups are present, all WALs are deleted, tables are removed
-     * from continuous backup metadata, and the associated replication peer is disabled.
+     * backup deletion.
      */
     private void cleanUpUnusedBackupWALs() throws IOException {
       Configuration conf = getConf() != null ? getConf() : HBaseConfiguration.create();
@@ -908,8 +904,7 @@ private void cleanUpUnusedBackupWALs() throws IOException {
         return;
       }
 
-      try (Admin admin = conn.getAdmin();
-        BackupSystemTable sysTable = new BackupSystemTable(conn)) {
+      try (BackupSystemTable sysTable = new BackupSystemTable(conn)) {
         // Get list of tables under continuous backup
         Map<TableName, Long> continuousBackupTables = sysTable.getContinuousBackupTableSet();
         if (continuousBackupTables.isEmpty()) {
@@ -920,15 +915,7 @@ private void cleanUpUnusedBackupWALs() throws IOException {
         // Find the earliest timestamp after which WALs are still needed
         long cutoffTimestamp = determineWALCleanupCutoffTime(sysTable);
         if (cutoffTimestamp == 0) {
-          // No full backup exists. PITR cannot function without a base full backup.
-          // Clean up all WALs, remove tables from backup metadata, and disable the replication
-          // peer.
-          System.out
-            .println("No full backups found. Cleaning up all WALs and disabling replication peer.");
-
-          disableContinuousBackupReplicationPeer(admin);
-          removeAllTablesFromContinuousBackup(sysTable);
-          deleteAllBackupWALFiles(conf, backupWalDir);
+          System.err.println("ERROR: No valid full backup found. Skipping WAL cleanup.");
           return;
         }
 
@@ -947,8 +934,7 @@ private void cleanUpUnusedBackupWALs() throws IOException {
      * @return cutoff timestamp or 0 if not found
      */
     long determineWALCleanupCutoffTime(BackupSystemTable sysTable) throws IOException {
-      List<BackupInfo> backupInfos =
-        sysTable.getBackupHistory(withState(BackupInfo.BackupState.COMPLETE));
+      List<BackupInfo> backupInfos = sysTable.getBackupInfos(BackupState.COMPLETE);
       Collections.reverse(backupInfos); // Start from oldest
 
       for (BackupInfo backupInfo : backupInfos) {
@@ -959,16 +945,6 @@ long determineWALCleanupCutoffTime(BackupSystemTable sysTable) throws IOExceptio
       return 0;
     }
 
-    private void disableContinuousBackupReplicationPeer(Admin admin) throws IOException {
-      for (ReplicationPeerDescription peer : admin.listReplicationPeers()) {
-        if (peer.getPeerId().equals(CONTINUOUS_BACKUP_REPLICATION_PEER) && peer.isEnabled()) {
-          admin.disableReplicationPeer(CONTINUOUS_BACKUP_REPLICATION_PEER);
-          System.out.println("Disabled replication peer: " + CONTINUOUS_BACKUP_REPLICATION_PEER);
-          break;
-        }
-      }
-    }
-
     /**
      * Updates the start time for continuous backups if older than cutoff timestamp.
      * @param sysTable        Backup system table
@@ -991,49 +967,6 @@ void updateBackupTableStartTimes(BackupSystemTable sysTable, long cutoffTimestam
       }
     }
 
-    private void removeAllTablesFromContinuousBackup(BackupSystemTable sysTable)
-      throws IOException {
-      Map<TableName, Long> allTables = sysTable.getContinuousBackupTableSet();
-      if (!allTables.isEmpty()) {
-        sysTable.removeContinuousBackupTableSet(allTables.keySet());
-        System.out.println("Removed all tables from continuous backup metadata.");
-      }
-    }
-
-    private void deleteAllBackupWALFiles(Configuration conf, String backupWalDir)
-      throws IOException {
-      try {
-        BackupFileSystemManager manager =
-          new BackupFileSystemManager(CONTINUOUS_BACKUP_REPLICATION_PEER, conf, backupWalDir);
-        FileSystem fs = manager.getBackupFs();
-        Path walDir = manager.getWalsDir();
-        Path bulkloadDir = manager.getBulkLoadFilesDir();
-
-        // Delete contents under WAL directory
-        if (fs.exists(walDir)) {
-          FileStatus[] walContents = fs.listStatus(walDir);
-          for (FileStatus item : walContents) {
-            fs.delete(item.getPath(), true); // recursive delete of each child
-          }
-          System.out.println("Deleted all contents under WAL directory: " + walDir);
-        }
-
-        // Delete contents under bulk load directory
-        if (fs.exists(bulkloadDir)) {
-          FileStatus[] bulkContents = fs.listStatus(bulkloadDir);
-          for (FileStatus item : bulkContents) {
-            fs.delete(item.getPath(), true); // recursive delete of each child
-          }
-          System.out.println("Deleted all contents under Bulk Load directory: " + bulkloadDir);
-        }
-
-      } catch (IOException e) {
-        System.out.println("WARNING: Failed to delete contents under backup directories: "
-          + backupWalDir + ". Error: " + e.getMessage());
-        throw e;
-      }
-    }
-
     /**
      * Cleans up old WAL and bulk-loaded files based on the determined cutoff timestamp.
      */
@@ -1072,15 +1005,13 @@ void deleteOldWALFiles(Configuration conf, String backupWalDir, long cutoffTime)
           if (dayStart + ONE_DAY_IN_MILLISECONDS - 1 < cutoffTime) {
             System.out.println("Deleting outdated WAL directory: " + dirPath);
             fs.delete(dirPath, true);
-            Path bulkloadPath = new Path(bulkloadDir, dirName);
-            System.out.println("Deleting corresponding bulk-load directory: " + bulkloadPath);
-            fs.delete(bulkloadPath, true);
+            fs.delete(new Path(bulkloadDir, dirName), true);
           }
         } catch (ParseException e) {
           System.out.println("WARNING: Failed to parse directory name '" + dirName
             + "'. Skipping. Error: " + e.getMessage());
         } catch (IOException e) {
-          System.err.println("WARNING: Failed to delete directory '" + dirPath
+          System.out.println("WARNING: Failed to delete directory '" + dirPath
             + "'. Skipping. Error: " + e.getMessage());
         }
       }

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java

@@ -1372,7 +1372,7 @@ private Delete createDeleteForIncrBackupTableSet(String backupRoot) {
   private Delete createDeleteForContinuousBackupTableSet(Set<TableName> tables) {
     Delete delete = new Delete(rowkey(CONTINUOUS_BACKUP_SET));
     for (TableName tableName : tables) {
-      delete.addColumn(META_FAMILY, Bytes.toBytes(tableName.getNameAsString()));
+      delete.addColumns(META_FAMILY, Bytes.toBytes(tableName.getNameAsString()));
     }
     return delete;
   }

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/DefaultPitrRestoreHandler.java

@@ -17,8 +17,6 @@
  */
 package org.apache.hadoop.hbase.backup.impl;
 
-import static org.apache.hadoop.hbase.backup.BackupInfo.withState;
-
 import java.io.IOException;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -50,7 +48,7 @@ public DefaultPitrRestoreHandler(Connection conn, PointInTimeRestoreRequest requ
   protected List<PitrBackupMetadata> getBackupMetadata(PointInTimeRestoreRequest request)
     throws IOException {
     try (BackupSystemTable table = new BackupSystemTable(conn)) {
-      return table.getBackupHistory(withState(BackupInfo.BackupState.COMPLETE)).stream()
+      return table.getBackupInfos(BackupInfo.BackupState.COMPLETE).stream()
         .map(BackupInfoAdapter::new).collect(Collectors.toList());
     }
   }