HBASE-29255: Integrate backup WAL cleanup logic with the delete command (#7007)

* Store bulkload files in daywise bucket as well

* Integrate backup WAL cleanup logic with the delete command

* address the review comments

* address the review comments

* address the review comments

* add more unit tests to cover all cases

* address the review comments

Author: vinayakphegde

hbase-backup/pom.xml

@@ -163,6 +163,11 @@
       <artifactId>junit</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/impl/BackupCommands.java

@@ -18,6 +18,8 @@
 package org.apache.hadoop.hbase.backup.impl;
 
 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;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_CONTINUOUS_BACKUP_PITR_WINDOW_DAYS;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_BACKUP_LIST_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_BANDWIDTH;
@@ -47,18 +49,26 @@
 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.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TimeZone;
 import java.util.concurrent.TimeUnit;
 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;
@@ -71,6 +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.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;
@@ -80,6 +91,7 @@
 import org.apache.yetus.audience.InterfaceAudience;
 
 import org.apache.hbase.thirdparty.com.google.common.base.Splitter;
+import org.apache.hbase.thirdparty.com.google.common.base.Strings;
 import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
 import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
 import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter;
@@ -649,6 +661,8 @@ public void execute() throws IOException {
       } else if (cmdline.hasOption(OPTION_LIST)) {
         executeDeleteListOfBackups(cmdline, isForceDelete);
       }
+
+      cleanUpUnusedBackupWALs();
     }
 
     private void executeDeleteOlderThan(CommandLine cmdline, boolean isForceDelete)
@@ -876,6 +890,139 @@ private boolean canAnyOtherBackupCover(List<BackupInfo> allBackups, BackupInfo c
       return false;
     }
 
+    /**
+     * Cleans up Write-Ahead Logs (WALs) that are no longer required for PITR after a successful
+     * backup deletion.
+     */
+    private void cleanUpUnusedBackupWALs() throws IOException {
+      Configuration conf = getConf() != null ? getConf() : HBaseConfiguration.create();
+      String backupWalDir = conf.get(CONF_CONTINUOUS_BACKUP_WAL_DIR);
+
+      if (Strings.isNullOrEmpty(backupWalDir)) {
+        System.out.println("No WAL directory specified for continuous backup. Skipping cleanup.");
+        return;
+      }
+
+      try (BackupSystemTable sysTable = new BackupSystemTable(conn)) {
+        // Get list of tables under continuous backup
+        Map<TableName, Long> continuousBackupTables = sysTable.getContinuousBackupTableSet();
+        if (continuousBackupTables.isEmpty()) {
+          System.out.println("No continuous backups configured. Skipping WAL cleanup.");
+          return;
+        }
+
+        // Find the earliest timestamp after which WALs are still needed
+        long cutoffTimestamp = determineWALCleanupCutoffTime(sysTable);
+        if (cutoffTimestamp == 0) {
+          System.err.println("ERROR: No valid full backup found. Skipping WAL cleanup.");
+          return;
+        }
+
+        // Update metadata before actual cleanup to avoid inconsistencies
+        updateBackupTableStartTimes(sysTable, cutoffTimestamp);
+
+        // Delete WAL files older than cutoff timestamp
+        deleteOldWALFiles(conf, backupWalDir, cutoffTimestamp);
+
+      }
+    }
+
+    /**
+     * Determines the cutoff time for cleaning WAL files.
+     * @param sysTable Backup system table
+     * @return cutoff timestamp or 0 if not found
+     */
+    long determineWALCleanupCutoffTime(BackupSystemTable sysTable) throws IOException {
+      List<BackupInfo> backupInfos = sysTable.getBackupInfos(BackupState.COMPLETE);
+      Collections.reverse(backupInfos); // Start from oldest
+
+      for (BackupInfo backupInfo : backupInfos) {
+        if (BackupType.FULL.equals(backupInfo.getType())) {
+          return backupInfo.getStartTs();
+        }
+      }
+      return 0;
+    }
+
+    /**
+     * Updates the start time for continuous backups if older than cutoff timestamp.
+     * @param sysTable        Backup system table
+     * @param cutoffTimestamp Timestamp before which WALs are no longer needed
+     */
+    void updateBackupTableStartTimes(BackupSystemTable sysTable, long cutoffTimestamp)
+      throws IOException {
+
+      Map<TableName, Long> backupTables = sysTable.getContinuousBackupTableSet();
+      Set<TableName> tablesToUpdate = new HashSet<>();
+
+      for (Map.Entry<TableName, Long> entry : backupTables.entrySet()) {
+        if (entry.getValue() < cutoffTimestamp) {
+          tablesToUpdate.add(entry.getKey());
+        }
+      }
+
+      if (!tablesToUpdate.isEmpty()) {
+        sysTable.updateContinuousBackupTableSet(tablesToUpdate, cutoffTimestamp);
+      }
+    }
+
+    /**
+     * Cleans up old WAL and bulk-loaded files based on the determined cutoff timestamp.
+     */
+    void deleteOldWALFiles(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);
+      dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
+
+      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, dirName), 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(DELETE_CMD_USAGE);

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

@@ -1056,6 +1056,32 @@ 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 {
+    if (tablesToUpdate == null || tablesToUpdate.isEmpty()) {
+      LOG.warn("No tables provided for updating start timestamps.");
+      return;
+    }
+
+    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());
+    }
+  }
+
   /**
    * Removes tables from the global continuous backup set. Only removes entries that currently exist
    * in the backup system table.

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

@@ -23,6 +23,7 @@
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -304,7 +305,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,
@@ -314,9 +315,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();
@@ -376,17 +375,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,
@@ -407,6 +410,15 @@ 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);
+    dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
+    return dateFormat.format(new Date(dayInMillis));
+  }
+
   private Path getBulkLoadFileStagingPath(Path relativePathFromNamespace) throws IOException {
     FileSystem rootFs = CommonFSUtils.getRootDirFileSystem(conf);
     Path rootDir = CommonFSUtils.getRootDir(conf);