HBASE-29209: Implement Backup Cleanup Command to Remove Older WALs Not Required for PITR of Any Table

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

@@ -120,6 +120,8 @@ private int parseAndRun(String[] args) throws IOException {
       type = BackupCommand.REPAIR;
     } else if (BackupCommand.MERGE.name().equalsIgnoreCase(cmd)) {
       type = BackupCommand.MERGE;
+    } else if (BackupCommand.CLEANUP.name().equalsIgnoreCase(cmd)) {
+      type = BackupCommand.CLEANUP;
     } else {
       System.out.println("Unsupported command for backup: " + cmd);
       printToolUsage();

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

@@ -153,6 +153,7 @@ enum BackupCommand {
     SET_DELETE,
     SET_DESCRIBE,
     SET_LIST,
-    REPAIR
+    REPAIR,
+    CLEANUP
   }
 }

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

@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hbase.backup.impl;
 
+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;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_BACKUP_LIST_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_BANDWIDTH;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_BANDWIDTH_DESC;
@@ -43,14 +45,22 @@
 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.replication.ContinuousBackupReplicationEndpoint.DATE_FORMAT;
+import static org.apache.hadoop.hbase.backup.replication.ContinuousBackupReplicationEndpoint.ONE_DAY_IN_MILLISECONDS;
 
 import java.io.IOException;
 import java.net.URI;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseConfiguration;
@@ -63,6 +73,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.replication.BackupFileSystemManager;
 import org.apache.hadoop.hbase.backup.util.BackupSet;
 import org.apache.hadoop.hbase.backup.util.BackupUtils;
 import org.apache.hadoop.hbase.client.Connection;
@@ -117,6 +128,8 @@ public final class BackupCommands {
 
   public static final String REPAIR_CMD_USAGE = "Usage: hbase backup repair\n";
 
+  public static final String CLEANUP_CMD_USAGE = "Usage: hbase backup cleanup\n";
+
   public static final String SET_CMD_USAGE = "Usage: hbase backup set COMMAND [name] [tables]\n"
     + "  name            Backup set name\n" + "  tables          Comma separated list of tables.\n"
     + "COMMAND is one of:\n" + "  add             add tables to a set, create a set if needed\n"
@@ -245,6 +258,9 @@ public static Command createCommand(Configuration conf, BackupCommand type, Comm
       case MERGE:
         cmd = new MergeCommand(conf, cmdline);
         break;
+      case CLEANUP:
+        cmd = new CleanupCommand(conf, cmdline);
+        break;
       case HELP:
       default:
         cmd = new HelpCommand(conf, cmdline);
@@ -853,6 +869,188 @@ protected void printUsage() {
     }
   }
 
+  /**
+   * The {@code CleanupCommand} class is responsible for removing Write-Ahead Log (WAL) and
+   * bulk-loaded files that are no longer needed for Point-in-Time Recovery (PITR).
+   * <p>
+   * The cleanup process follows these steps:
+   * <ol>
+   * <li>Identify the oldest full backup and its start timestamp.</li>
+   * <li>Delete WAL files older than this timestamp, as they are no longer usable for PITR with any
+   * backup.</li>
+   * </ol>
+   */
+  public static class CleanupCommand extends Command {
+    CleanupCommand(Configuration conf, CommandLine cmdline) {
+      super(conf);
+      this.cmdline = cmdline;
+    }
+
+    @Override
+    public void execute() throws IOException {
+      super.execute();
+
+      // Validate input arguments
+      validateArguments();
+
+      Configuration conf = getConf() != null ? getConf() : HBaseConfiguration.create();
+      String backupWalDir = conf.get(CONF_CONTINUOUS_BACKUP_WAL_DIR);
+
+      if (backupWalDir == null || backupWalDir.isEmpty()) {
+        System.out
+          .println("WAL Directory is not specified for continuous backup. Nothing to clean!");
+        return;
+      }
+
+      try (final Connection conn = ConnectionFactory.createConnection(conf);
+        final BackupSystemTable sysTable = new BackupSystemTable(conn)) {
+
+        // Retrieve tables that are part of continuous backup
+        Map<TableName, Long> continuousBackupTables = sysTable.getContinuousBackupTableSet();
+        if (continuousBackupTables.isEmpty()) {
+          System.out.println("Continuous Backup is not enabled for any tables. Nothing to clean!");
+          return;
+        }
+
+        // Determine the earliest timestamp before which WAL files can be deleted
+        long cleanupCutoffTimestamp = determineCleanupCutoffTime(sysTable, continuousBackupTables);
+        if (cleanupCutoffTimestamp == 0) {
+          System.err.println("ERROR: No valid full backup found. Cleanup aborted.");
+          return;
+        }
+
+        // Update the continuous backup table's start time to match the cutoff time *before* actual
+        // cleanup.
+        // This is safe because even if the WAL cleanup fails later, we won't be accessing data
+        // older than
+        // the cutoff timestamp, ensuring consistency in what the system considers valid for
+        // recovery.
+        //
+        // If we did this the other way around—cleaning up first and updating the table afterward—
+        // a failure between these two steps could leave us in an inconsistent state where some WALs
+        // are already deleted, but the backup metadata still references them.
+        updateContinuousBackupTablesStartTime(sysTable, cleanupCutoffTimestamp);
+
+        // Perform WAL file cleanup
+        cleanupOldWALFiles(conf, backupWalDir, cleanupCutoffTimestamp);
+      }
+    }
+
+    /**
+     * Fetches the continuous backup tables from the system table and updates their start timestamps
+     * if the current start time is earlier than the given cutoff timestamp.
+     * @param sysTable               The backup system table from which continuous backup tables are
+     *                               retrieved and updated.
+     * @param cleanupCutoffTimestamp The cutoff timestamp before which WAL files can be deleted.
+     * @throws IOException If an error occurs while accessing the system table.
+     */
+    private void updateContinuousBackupTablesStartTime(BackupSystemTable sysTable,
+      long cleanupCutoffTimestamp) throws IOException {
+      Map<TableName, Long> continuousBackupTables = sysTable.getContinuousBackupTableSet();
+
+      // Identify tables that need updating
+      Set<TableName> tablesToUpdate = new HashSet<>();
+      for (Map.Entry<TableName, Long> entry : continuousBackupTables.entrySet()) {
+        TableName table = entry.getKey();
+        long startTimestamp = entry.getValue();
+
+        if (startTimestamp < cleanupCutoffTimestamp) {
+          tablesToUpdate.add(table);
+        }
+      }
+
+      // If no tables require updates, exit early
+      if (tablesToUpdate.isEmpty()) {
+        return;
+      }
+
+      // Perform the actual update in the system table
+      sysTable.updateContinuousBackupTableSet(tablesToUpdate, cleanupCutoffTimestamp);
+    }
+
+    private void validateArguments() throws IOException {
+      String[] args = cmdline == null ? null : cmdline.getArgs();
+      if (args != null && args.length > 1) {
+        System.err.println("ERROR: wrong number of arguments: " + args.length);
+        printUsage();
+        throw new IOException(INCORRECT_USAGE);
+      }
+    }
+
+    private long determineCleanupCutoffTime(BackupSystemTable sysTable,
+      Map<TableName, Long> backupTables) throws IOException {
+      List<BackupInfo> backupInfos = sysTable.getBackupInfos(BackupState.COMPLETE);
+      Collections.reverse(backupInfos); // Process from oldest to latest
+
+      for (BackupInfo backupInfo : backupInfos) {
+        if (BackupType.FULL.equals(backupInfo.getType())) {
+          return backupInfo.getStartTs();
+        }
+      }
+      return 0;
+    }
+
+    /**
+     * Cleans up old WAL and bulk-loaded files based on the determined cutoff timestamp.
+     */
+    private void cleanupOldWALFiles(Configuration conf, String backupWalDir, long cutoffTime)
+      throws IOException {
+      System.out.println("Starting WAL cleanup in backup directory: " + backupWalDir
+        + " with cutoff time: " + cutoffTime);
+
+      BackupFileSystemManager manager =
+        new BackupFileSystemManager(CONTINUOUS_BACKUP_REPLICATION_PEER, conf, backupWalDir);
+      FileSystem fs = manager.getBackupFs();
+      Path walDir = manager.getWalsDir();
+      Path bulkloadDir = manager.getBulkLoadFilesDir();
+      SimpleDateFormat dateFormat = new SimpleDateFormat(DATE_FORMAT);
+
+      System.out.println("Listing directories under: " + walDir);
+
+      FileStatus[] directories = fs.listStatus(walDir);
+
+      for (FileStatus dirStatus : directories) {
+        if (!dirStatus.isDirectory()) {
+          continue; // Skip files, we only want directories
+        }
+
+        Path dirPath = dirStatus.getPath();
+        String dirName = dirPath.getName();
+
+        try {
+          long dayStart = parseDayDirectory(dirName, dateFormat);
+          System.out
+            .println("Checking WAL directory: " + dirName + " (Start Time: " + dayStart + ")");
+
+          // If WAL files of that day are older than cutoff time, delete them
+          if (dayStart + ONE_DAY_IN_MILLISECONDS - 1 < cutoffTime) {
+            System.out.println("Deleting outdated WAL directory: " + dirPath);
+            fs.delete(dirPath, true);
+            fs.delete(new Path(bulkloadDir, dirPath.getName()), true);
+          }
+        } catch (ParseException e) {
+          System.out.println("WARNING: Failed to parse directory name '" + dirName
+            + "'. Skipping. Error: " + e.getMessage());
+        } catch (IOException e) {
+          System.out.println("WARNING: Failed to delete directory '" + dirPath
+            + "'. Skipping. Error: " + e.getMessage());
+        }
+      }
+
+      System.out.println("Completed WAL cleanup for backup directory: " + backupWalDir);
+    }
+
+    private long parseDayDirectory(String dayDir, SimpleDateFormat dateFormat)
+      throws ParseException {
+      return dateFormat.parse(dayDir).getTime();
+    }
+
+    @Override
+    protected void printUsage() {
+      System.out.println(CLEANUP_CMD_USAGE);
+    }
+  }
+
   public static class MergeCommand extends Command {
     MergeCommand(Configuration conf, CommandLine cmdline) {
       super(conf);

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

@@ -1050,6 +1050,27 @@ public void addContinuousBackupTableSet(Set<TableName> tables, long startTimesta
     }
   }
 
+  /**
+   * Updates the system table with the new start timestamps for continuous backup tables.
+   * @param tablesToUpdate    The set of tables that need their start timestamps updated.
+   * @param newStartTimestamp The new start timestamp to be set.
+   */
+  public void updateContinuousBackupTableSet(Set<TableName> tablesToUpdate, long newStartTimestamp)
+    throws IOException {
+    try (Table table = connection.getTable(tableName)) {
+      Put put = new Put(rowkey(CONTINUOUS_BACKUP_SET));
+
+      for (TableName tableName : tablesToUpdate) {
+        put.addColumn(BackupSystemTable.META_FAMILY, Bytes.toBytes(tableName.getNameAsString()),
+          Bytes.toBytes(newStartTimestamp));
+      }
+
+      table.put(put);
+      LOG.info("Successfully updated start timestamps for {} tables in the backup system table.",
+        tablesToUpdate.size());
+    }
+  }
+
   /**
    * Deletes incremental backup set for a backup destination
    * @param backupRoot backup root

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/replication/ContinuousBackupReplicationEndpoint.java

@@ -271,7 +271,7 @@ private void backupWalEntries(long day, List<WAL.Entry> walEntries) throws IOExc
         walWriter.append(entry);
       }
       walWriter.sync(true);
-      uploadBulkLoadFiles(bulkLoadFiles);
+      uploadBulkLoadFiles(day, bulkLoadFiles);
     } catch (UncheckedIOException e) {
       String errorMsg = Utils.logPeerId(peerId) + " Failed to get or create WAL Writer for " + day;
       LOG.error("{} Backup failed for day {}. Error: {}", Utils.logPeerId(peerId), day,
@@ -281,9 +281,7 @@ private void backupWalEntries(long day, List<WAL.Entry> walEntries) throws IOExc
   }
 
   private FSHLogProvider.Writer createWalWriter(long dayInMillis) {
-    // Convert dayInMillis to "yyyy-MM-dd" format
-    SimpleDateFormat dateFormat = new SimpleDateFormat(DATE_FORMAT);
-    String dayDirectoryName = dateFormat.format(new Date(dayInMillis));
+    String dayDirectoryName = formatToDateString(dayInMillis);
 
     FileSystem fs = backupFileSystemManager.getBackupFs();
     Path walsDir = backupFileSystemManager.getWalsDir();
@@ -343,17 +341,21 @@ private void close() {
     }
   }
 
-  private void uploadBulkLoadFiles(List<Path> bulkLoadFiles) throws IOException {
+  private void uploadBulkLoadFiles(long dayInMillis, List<Path> bulkLoadFiles) throws IOException {
     LOG.debug("{} Starting upload of {} bulk load files", Utils.logPeerId(peerId),
       bulkLoadFiles.size());
 
     if (LOG.isTraceEnabled()) {
       LOG.trace("{} Bulk load files to upload: {}", Utils.logPeerId(peerId),
         bulkLoadFiles.stream().map(Path::toString).collect(Collectors.joining(", ")));
     }
+    String dayDirectoryName = formatToDateString(dayInMillis);
+    Path bulkloadDir = new Path(backupFileSystemManager.getBulkLoadFilesDir(), dayDirectoryName);
+    backupFileSystemManager.getBackupFs().mkdirs(bulkloadDir);
+
     for (Path file : bulkLoadFiles) {
       Path sourcePath = getBulkLoadFileStagingPath(file);
-      Path destPath = new Path(backupFileSystemManager.getBulkLoadFilesDir(), file);
+      Path destPath = new Path(bulkloadDir, file);
 
       try {
         LOG.debug("{} Copying bulk load file from {} to {}", Utils.logPeerId(peerId), sourcePath,
@@ -374,6 +376,14 @@ private void uploadBulkLoadFiles(List<Path> bulkLoadFiles) throws IOException {
     LOG.debug("{} Completed upload of bulk load files", Utils.logPeerId(peerId));
   }
 
+  /**
+   * Convert dayInMillis to "yyyy-MM-dd" format
+   */
+  private String formatToDateString(long dayInMillis) {
+    SimpleDateFormat dateFormat = new SimpleDateFormat(DATE_FORMAT);
+    return dateFormat.format(new Date(dayInMillis));
+  }
+
   private Path getBulkLoadFileStagingPath(Path relativePathFromNamespace) throws IOException {
     FileSystem rootFs = CommonFSUtils.getRootDirFileSystem(conf);
     Path rootDir = CommonFSUtils.getRootDir(conf);