HBASE-29029 Refactor BackupHFileCleaner + fix test

The TestBackupHFileCleaner test was broken, as it used
a different API to register bulk loads than the API that
is actually used to register bulk loads during backups.
The test also incorrectly closed the FS of the HBaseTestingUtil,
causing this test to block for about 5 minutes during shutdown.

Both the test and BackupHFileCleaner itself were overly convoluted
and are cleaned up.

Author: DieterDP-ng

hbase-backup/pom.xml

@@ -112,6 +112,12 @@
       <scope>test</scope>
     </dependency>
     <!-- General dependencies -->
+    <dependency>
+      <groupId>com.github.stephenc.findbugs</groupId>
+      <artifactId>findbugs-annotations</artifactId>
+      <scope>compile</scope>
+      <optional>true</optional>
+    </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>

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

@@ -18,19 +18,18 @@
 package org.apache.hadoop.hbase.backup;
 
 import java.io.IOException;
-import java.util.ArrayList;
 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.NotImplementedException;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Abortable;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
+import org.apache.hadoop.hbase.backup.impl.BulkLoad;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.master.cleaner.BaseHFileCleanerDelegate;
@@ -42,106 +41,81 @@
 import org.apache.hbase.thirdparty.com.google.common.collect.Iterables;
 
 /**
- * Implementation of a file cleaner that checks if an hfile is still referenced by backup before
- * deleting it from hfile archive directory.
+ * File cleaner that prevents deletion of HFiles that are still required by future incremental
+ * backups.
+ * <p>
+ * Bulk loaded HFiles that are needed by future updates are stored in the backup system table.
  */
 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
 public class BackupHFileCleaner extends BaseHFileCleanerDelegate implements Abortable {
   private static final Logger LOG = LoggerFactory.getLogger(BackupHFileCleaner.class);
+
   private boolean stopped = false;
-  private boolean aborted;
-  private Configuration conf;
+  private boolean aborted = false;
   private Connection connection;
-  private long prevReadFromBackupTbl = 0, // timestamp of most recent read from backup:system table
-      secondPrevReadFromBackupTbl = 0; // timestamp of 2nd most recent read from backup:system table
-  // used by unit test to skip reading backup:system
-  private boolean checkForFullyBackedUpTables = true;
-  private List<TableName> fullyBackedUpTables = null;
-
-  private Set<String> getFilenameFromBulkLoad(Map<byte[], List<Path>>[] maps) {
-    Set<String> filenames = new HashSet<>();
-    for (Map<byte[], List<Path>> map : maps) {
-      if (map == null) {
-        continue;
-      }
-
-      for (List<Path> paths : map.values()) {
-        for (Path p : paths) {
-          filenames.add(p.getName());
-        }
-      }
-    }
-    return filenames;
-  }
-
-  private Set<String> loadHFileRefs(List<TableName> tableList) throws IOException {
-    if (connection == null) {
-      connection = ConnectionFactory.createConnection(conf);
-    }
-    try (BackupSystemTable tbl = new BackupSystemTable(connection)) {
-      Map<byte[], List<Path>>[] res = tbl.readBulkLoadedFiles(null, tableList);
-      secondPrevReadFromBackupTbl = prevReadFromBackupTbl;
-      prevReadFromBackupTbl = EnvironmentEdgeManager.currentTime();
-      return getFilenameFromBulkLoad(res);
-    }
-  }
-
-  @InterfaceAudience.Private
-  void setCheckForFullyBackedUpTables(boolean b) {
-    checkForFullyBackedUpTables = b;
-  }
+  // timestamp of most recent read from backup system table
+  private long prevReadFromBackupTbl = 0;
+  // timestamp of 2nd most recent read from backup system table
+  private long secondPrevReadFromBackupTbl = 0;
 
   @Override
   public Iterable<FileStatus> getDeletableFiles(Iterable<FileStatus> files) {
-    if (conf == null) {
-      return files;
+    if (stopped) {
+      return Collections.emptyList();
     }
-    // obtain the Set of TableName's which have been fully backed up
-    // so that we filter BulkLoad to be returned from server
-    if (checkForFullyBackedUpTables) {
-      if (connection == null) {
-        return files;
-      }
 
-      try (BackupSystemTable tbl = new BackupSystemTable(connection)) {
-        fullyBackedUpTables = new ArrayList<>(tbl.getTablesIncludedInBackups());
-      } catch (IOException ioe) {
-        LOG.error("Failed to get tables which have been fully backed up, skipping checking", ioe);
-        return Collections.emptyList();
+    // We use filenames because the HFile will have been moved to the archive since it
+    // was registered.
+    final Set<String> hfileFilenames = new HashSet<>();
+    try (BackupSystemTable tbl = new BackupSystemTable(connection)) {
+      Set<TableName> tablesIncludedInBackups = fetchFullyBackedUpTables(tbl);
+      for (BulkLoad bulkLoad : tbl.readBulkloadRows(tablesIncludedInBackups)) {
+        hfileFilenames.add(new Path(bulkLoad.getHfilePath()).getName());
       }
-      Collections.sort(fullyBackedUpTables);
-    }
-    final Set<String> hfileRefs;
-    try {
-      hfileRefs = loadHFileRefs(fullyBackedUpTables);
+      LOG.debug("Found {} unique HFile filenames registered as bulk loads.", hfileFilenames.size());
     } catch (IOException ioe) {
-      LOG.error("Failed to read hfile references, skipping checking deletable files", ioe);
+      LOG.error(
+        "Failed to read registered bulk load references from backup system table, marking all files as non-deletable.",
+        ioe);
       return Collections.emptyList();
     }
-    Iterable<FileStatus> deletables = Iterables.filter(files, file -> {
-      // If the file is recent, be conservative and wait for one more scan of backup:system table
+
+    secondPrevReadFromBackupTbl = prevReadFromBackupTbl;
+    prevReadFromBackupTbl = EnvironmentEdgeManager.currentTime();
+
+    return Iterables.filter(files, file -> {
+      // If the file is recent, be conservative and wait for one more scan of the bulk loads
       if (file.getModificationTime() > secondPrevReadFromBackupTbl) {
+        LOG.debug("Preventing deletion due to timestamp: {}", file.getPath().toString());
         return false;
       }
+      // A file can be deleted if it is not registered as a backup bulk load.
       String hfile = file.getPath().getName();
-      boolean foundHFileRef = hfileRefs.contains(hfile);
-      return !foundHFileRef;
+      if (hfileFilenames.contains(hfile)) {
+        LOG.debug("Preventing deletion due to bulk load registration in backup system table: {}",
+          file.getPath().toString());
+        return false;
+      } else {
+        LOG.debug("OK to delete: {}", file.getPath().toString());
+        return true;
+      }
     });
-    return deletables;
+  }
+
+  protected Set<TableName> fetchFullyBackedUpTables(BackupSystemTable tbl) throws IOException {
+    return tbl.getTablesIncludedInBackups();
   }
 
   @Override
   public boolean isFileDeletable(FileStatus fStat) {
-    // work is done in getDeletableFiles()
-    return true;
+    throw new NotImplementedException("This method should not be called");
   }
 
   @Override
   public void setConf(Configuration config) {
-    this.conf = config;
     this.connection = null;
     try {
-      this.connection = ConnectionFactory.createConnection(conf);
+      this.connection = ConnectionFactory.createConnection(config);
     } catch (IOException ioe) {
       LOG.error("Couldn't establish connection", ioe);
     }
@@ -156,7 +130,7 @@ public void stop(String why) {
       try {
         this.connection.close();
       } catch (IOException ioe) {
-        LOG.debug("Got " + ioe + " when closing connection");
+        LOG.debug("Got IOException when closing connection", ioe);
       }
     }
     this.stopped = true;
@@ -169,7 +143,7 @@ public boolean isStopped() {
 
   @Override
   public void abort(String why, Throwable e) {
-    LOG.warn("Aborting ReplicationHFileCleaner because " + why, e);
+    LOG.warn("Aborting ReplicationHFileCleaner because {}", why, e);
     this.aborted = true;
     stop(why);
   }