HBASE-29003 Proper bulk load tracking

The HBase backup mechanism keeps track of which HFiles
were bulk loaded, so they can be included in incremental
backups.

Before this ticket, these bulk load records were only
deleted when an incremental backup is created. This
commit adds 2 more locations:

1) after a full backup. Since a full backup already
captures all data, this meant that unnecessary HFiles
were included in the next incremental backup.

2) after a table delete/truncate. Previously, if an
HFile was loaded before a table was cleared, the next
incremental backup would effectively still include the
HFile. This lead to incorrect data being restored.

This commit also completely refactors & simplifies the
test for this functionality.

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

@@ -19,20 +19,29 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.impl.BackupManager;
 import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
+import org.apache.hadoop.hbase.backup.impl.BulkLoad;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.coprocessor.MasterCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.MasterCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.MasterObserver;
 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
@@ -42,18 +51,26 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
+
 /**
  * An Observer to facilitate backup operations
  */
 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
-public class BackupObserver implements RegionCoprocessor, RegionObserver {
+public class BackupObserver
+  implements RegionCoprocessor, RegionObserver, MasterCoprocessor, MasterObserver {
   private static final Logger LOG = LoggerFactory.getLogger(BackupObserver.class);
 
   @Override
   public Optional<RegionObserver> getRegionObserver() {
     return Optional.of(this);
   }
 
+  @Override
+  public Optional<MasterObserver> getMasterObserver() {
+    return Optional.of(this);
+  }
+
   @Override
   public void postBulkLoadHFile(ObserverContext<? extends RegionCoprocessorEnvironment> ctx,
     List<Pair<byte[], String>> stagingFamilyPaths, Map<byte[], List<Path>> finalPaths)
@@ -106,4 +123,62 @@ private void registerBulkLoad(ObserverContext<? extends RegionCoprocessorEnviron
       }
     }
   }
+
+  @Override
+  public void postDeleteTable(ObserverContext<MasterCoprocessorEnvironment> ctx,
+    TableName tableName) throws IOException {
+    Configuration cfg = ctx.getEnvironment().getConfiguration();
+    if (!BackupManager.isBackupEnabled(cfg)) {
+      LOG.debug("Skipping postDeleteTable hook since backup is disabled");
+      return;
+    }
+    deleteBulkLoads(cfg, tableName, (ignored) -> true);
+  }
+
+  @Override
+  public void postTruncateTable(ObserverContext<MasterCoprocessorEnvironment> ctx,
+    TableName tableName) throws IOException {
+    Configuration cfg = ctx.getEnvironment().getConfiguration();
+    if (!BackupManager.isBackupEnabled(cfg)) {
+      LOG.debug("Skipping postTruncateTable hook since backup is disabled");
+      return;
+    }
+    deleteBulkLoads(cfg, tableName, (ignored) -> true);
+  }
+
+  @Override
+  public void postModifyTable(final ObserverContext<MasterCoprocessorEnvironment> ctx,
+    final TableName tableName, TableDescriptor oldDescriptor, TableDescriptor currentDescriptor)
+    throws IOException {
+    Configuration cfg = ctx.getEnvironment().getConfiguration();
+    if (!BackupManager.isBackupEnabled(cfg)) {
+      LOG.debug("Skipping postModifyTable hook since backup is disabled");
+      return;
+    }
+
+    Set<String> oldFamilies = Arrays.stream(oldDescriptor.getColumnFamilies())
+      .map(ColumnFamilyDescriptor::getNameAsString).collect(Collectors.toSet());
+    Set<String> newFamilies = Arrays.stream(currentDescriptor.getColumnFamilies())
+      .map(ColumnFamilyDescriptor::getNameAsString).collect(Collectors.toSet());
+
+    Set<String> removedFamilies = Sets.difference(oldFamilies, newFamilies);
+    if (!removedFamilies.isEmpty()) {
+      Predicate<BulkLoad> filter = bulkload -> removedFamilies.contains(bulkload.getColumnFamily());
+      deleteBulkLoads(cfg, tableName, filter);
+    }
+  }
+
+  /**
+   * Deletes all bulk load entries for the given table, matching the provided predicate.
+   */
+  private void deleteBulkLoads(Configuration config, TableName tableName,
+    Predicate<BulkLoad> filter) throws IOException {
+    try (Connection connection = ConnectionFactory.createConnection(config);
+      BackupSystemTable tbl = new BackupSystemTable(connection)) {
+      List<BulkLoad> bulkLoads = tbl.readBulkloadRows(List.of(tableName));
+      List<byte[]> rowsToDelete =
+        bulkLoads.stream().filter(filter).map(BulkLoad::getRowKey).toList();
+      tbl.deleteBulkLoadedRows(rowsToDelete);
+    }
+  }
 }

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.backup;
 
 import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**
@@ -105,8 +106,9 @@ public interface BackupRestoreConstants {
       + "org.apache.hadoop.hbase.backup.master.LogRollMasterProcedureManager\n"
       + "hbase.procedure.regionserver.classes=YOUR_CLASSES,"
       + "org.apache.hadoop.hbase.backup.regionserver.LogRollRegionServerProcedureManager\n"
-      + "hbase.coprocessor.region.classes=YOUR_CLASSES,"
-      + "org.apache.hadoop.hbase.backup.BackupObserver\n" + "and restart the cluster\n"
+      + CoprocessorHost.REGION_COPROCESSOR_CONF_KEY + "=YOUR_CLASSES,"
+      + BackupObserver.class.getSimpleName() + "\n" + CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY
+      + "=YOUR_CLASSES," + BackupObserver.class.getSimpleName() + "\nand restart the cluster\n"
       + "For more information please see http://hbase.apache.org/book.html#backuprestore\n";
   String ENABLE_BACKUP = "Backup is not enabled. To enable backup, " + "in hbase-site.xml, set:\n "
     + BACKUP_CONFIG_STRING;

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

@@ -146,13 +146,17 @@ public static void decorateRegionServerConfiguration(Configuration conf) {
         classes + "," + regionProcedureClass);
     }
     String coproc = conf.get(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY);
-    String regionObserverClass = BackupObserver.class.getName();
+    String observerClass = BackupObserver.class.getName();
     conf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
-      (coproc == null ? "" : coproc + ",") + regionObserverClass);
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("Added region procedure manager: {}. Added region observer: {}",
-        regionProcedureClass, regionObserverClass);
-    }
+      (coproc == null ? "" : coproc + ",") + observerClass);
+
+    String masterCoProc = conf.get(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY);
+    conf.set(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY,
+      (masterCoProc == null ? "" : masterCoProc + ",") + observerClass);
+
+    LOG.debug(
+      "Added region procedure manager: {}. Added region observer: {}. Added master observer: {}",
+      regionProcedureClass, observerClass, observerClass);
   }
 
   public static boolean isBackupEnabled(Configuration conf) {

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

@@ -411,25 +411,24 @@ public void registerBulkLoad(TableName tableName, byte[] region,
     try (BufferedMutator bufferedMutator = connection.getBufferedMutator(bulkLoadTableName)) {
       List<Put> puts = BackupSystemTable.createPutForBulkLoad(tableName, region, cfToHfilePath);
       bufferedMutator.mutate(puts);
-      LOG.debug("Written {} rows for bulk load of {}", puts.size(), tableName);
+      LOG.debug("Written {} rows for bulk load of table {}", puts.size(), tableName);
     }
   }
 
-  /*
-   * Removes rows recording bulk loaded hfiles from backup table
-   * @param lst list of table names
-   * @param rows the rows to be deleted
+  /**
+   * Removes entries from the table that tracks all bulk loaded hfiles.
+   * @param rows the row keys of the entries to be deleted
    */
   public void deleteBulkLoadedRows(List<byte[]> rows) throws IOException {
     try (BufferedMutator bufferedMutator = connection.getBufferedMutator(bulkLoadTableName)) {
-      List<Delete> lstDels = new ArrayList<>();
+      List<Delete> deletes = new ArrayList<>();
       for (byte[] row : rows) {
         Delete del = new Delete(row);
-        lstDels.add(del);
-        LOG.debug("orig deleting the row: " + Bytes.toString(row));
+        deletes.add(del);
+        LOG.debug("Deleting bulk load entry with key: {}", Bytes.toString(row));
       }
-      bufferedMutator.mutate(lstDels);
-      LOG.debug("deleted " + rows.size() + " original bulkload rows");
+      bufferedMutator.mutate(deletes);
+      LOG.debug("Deleted {} bulk load entries.", rows.size());
     }
   }
 
@@ -1522,16 +1521,6 @@ public static void deleteSnapshot(Connection conn) throws IOException {
     }
   }
 
-  public static List<Delete> createDeleteForOrigBulkLoad(List<TableName> lst) {
-    List<Delete> lstDels = new ArrayList<>(lst.size());
-    for (TableName table : lst) {
-      Delete del = new Delete(rowkey(BULK_LOAD_PREFIX, table.toString(), BLK_LD_DELIM));
-      del.addFamily(BackupSystemTable.META_FAMILY);
-      lstDels.add(del);
-    }
-    return lstDels;
-  }
-
   private Put createPutForDeleteOperation(String[] backupIdList) {
     byte[] value = Bytes.toBytes(StringUtils.join(backupIdList, ","));
     Put put = new Put(DELETE_OP_ROW);

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

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.BackupCopyJob;
@@ -152,6 +153,11 @@ public void execute() throws IOException {
       // the snapshot.
       LOG.info("Execute roll log procedure for full backup ...");
 
+      // Gather the bulk loads being tracked by the system, which can be deleted (since their data
+      // will be part of the snapshot being taken). We gather this list before taking the actual
+      // snapshots for the same reason as the log rolls.
+      List<BulkLoad> bulkLoadsToDelete = backupManager.readBulkloadRows(tableList);
+
       Map<String, String> props = new HashMap<>();
       props.put("backupRoot", backupInfo.getBackupRootDir());
       admin.execProcedure(LogRollMasterProcedureManager.ROLLLOG_PROCEDURE_SIGNATURE,
@@ -192,6 +198,9 @@ public void execute() throws IOException {
         BackupUtils.getMinValue(BackupUtils.getRSLogTimestampMins(newTableSetTimestampMap));
       backupManager.writeBackupStartCode(newStartCode);
 
+      backupManager
+        .deleteBulkLoadedRows(bulkLoadsToDelete.stream().map(BulkLoad::getRowKey).toList());
+
       // backup complete
       completeBackup(conn, backupInfo, BackupType.FULL, conf);
     } catch (Exception e) {