Add metrics for HDFS checksum and no-checksum bytes read

.blazar.yaml

@@ -0,0 +1,25 @@
+buildpack:
+  name: Blazar-Buildpack-Java-oss-fork
+
+env:
+  MAVEN_PHASE: "package assembly:single deploy"
+  HADOOP_DEP_VERSION: "3.3.6-hubspot-SNAPSHOT"
+  MAVEN_BUILD_ARGS: "-Phadoop-3.0 -Dhadoop.profile=3.0 -Dhadoop-three.version=$HADOOP_DEP_VERSION -Dgpg.skip=true -DskipTests -DdeployAtEnd -pl hbase-assembly -am -T1C"
+
+  # Below variables are generated in prepare_environment.sh.
+  # The build environment requires environment variables to be explicitly defined before they may
+  # be modified by the `write-build-env-var` utilty script to persist changes to an environment variable
+  # throughout a build
+  REPO_NAME: ""
+  SET_VERSION: ""
+  HBASE_VERSION: ""
+  PKG_RELEASE: ""
+  FULL_BUILD_VERSION: ""
+
+before:
+  - description: "Prepare build environment"
+    commands:
+      - $WORKSPACE/build-scripts/prepare_environment.sh
+
+provides:
+  - hbase

CLAUDE.md

@@ -0,0 +1,5 @@
+When in the hubspot-2.6 branch, or sub-branches of that branch, use mvn11 instead of mvn, and use the hadoop-3.0 maven profile
+
+When in the master branch, or sub-branches of that branch, use mvn17 instead of mvn, and do not use the hadoop-3.0 maven profile.
+
+Always run the spotless:apply maven target after making changes.

build-scripts/prepare_environment.sh

@@ -0,0 +1,97 @@
+#
+# Generates the appropriate environment vars so that we:
+# - build against the right version of hadoop, and properly set up maven
+# - generate the correct maven version based on the branches
+# - upload RPMs with the correct release based on the branch, and to the right yum repo
+#
+# Since we need to distribute .blazar.yaml to all sub-modules of the project, we define our constants once
+# in this script which can be re-used by every .blazar.yaml.
+#
+set -ex
+printenv
+
+# We base the expected main branch and resulting maven version for clients on the hbase minor version
+# The reason for this is hbase re-branches for each minor release (2.4, 2.5, 2.6, etc). At each re-branch
+# the histories diverge. So we'll need to create our own fork of each new minor release branch.
+# The convention is a fork named "hubspot-$minorVersion", and the maven coordinates "$minorVersion-hubspot-SNAPSHOT"
+MINOR_VERSION="2.6"
+MAIN_BRANCH="hubspot-${MINOR_VERSION}"
+
+#
+# Validate inputs from blazar
+#
+
+if [ -z "$WORKSPACE" ]; then
+    echo "Missing env var \$WORKSPACE"
+    exit 1
+fi
+if [ -z "$GIT_BRANCH" ]; then
+    echo "Missing env var \$GIT_BRANCH"
+    exit 1
+fi
+if [ -z "$BUILD_COMMAND_RC_FILE" ]; then
+    echo "Missing env var \$BUILD_COMMAND_RC_FILE"
+    exit 1
+fi
+
+#
+# Extract current hbase version from root pom.xml
+#
+
+# the pom.xml has an invalid xml namespace, so just remove that so xmllint can parse it.
+cat $WORKSPACE/pom.xml | sed '2 s/xmlns=".*"//g' > pom.xml.tmp
+HBASE_VERSION=$(echo "cat /project/properties/revision/text()" | xmllint --nocdata --shell pom.xml.tmp | sed '1d;$d')
+rm pom.xml.tmp
+
+# sanity check that we've got some that looks right. it wouldn't be the end of the world if we got it wrong, but
+# will help avoid confusion.
+if [[ ! "$HBASE_VERSION" =~ 2\.[0-9]+\.[0-9]+ ]]; then
+    echo "Unexpected HBASE_Version extracted from pom.xml. Got $HBASE_VERSION but expected a string like '2.4.3', with 3 numbers separated by decimals, the first number being 2."
+    exit 1
+fi
+
+#
+# Generate branch-specific env vars
+# We are going to generate the maven version and the RPM release here:
+# - For the maven version, we need to special case our main branch
+# - For RPM, we want our final version to be:
+#   main branch: {hbase_version}-hs.{build_number}.el6
+#   other branches: {hbase_version}-hs~{branch_name}.{build_number}.el6, where branch_name substitutes underscore for non-alpha-numeric characters
+#
+
+echo "Git branch $GIT_BRANCH. Detecting appropriate version override and RPM release."
+
+RELEASE="hs"
+
+if [[ "$GIT_BRANCH" = "$MAIN_BRANCH" ]]; then
+    SET_VERSION="${MINOR_VERSION}-hubspot-SNAPSHOT"
+    REPO_NAME="AnyLinuxVersion_hs-hbase"
+elif [[ "$GIT_BRANCH" != "hubspot" ]]; then
+    SET_VERSION="${MINOR_VERSION}-${GIT_BRANCH}-SNAPSHOT"
+    RELEASE="${RELEASE}~${GIT_BRANCH//[^[:alnum:]]/_}"
+    REPO_NAME="AnyLinuxVersion_hs-hbase-develop"
+else
+    echo "Invalid git branch $GIT_BRANCH"
+    exit 1
+fi
+
+RELEASE="${RELEASE}.${BUILD_NUMBER}"
+FULL_BUILD_VERSION="${HBASE_VERSION}-${RELEASE}"
+
+# SET_VERSION is not the most intuitive name, but it's required for set-maven-versions script
+write-build-env-var SET_VERSION "$SET_VERSION"
+write-build-env-var HBASE_VERSION "$HBASE_VERSION"
+write-build-env-var PKG_RELEASE "$RELEASE"
+write-build-env-var FULL_BUILD_VERSION "$FULL_BUILD_VERSION"
+write-build-env-var REPO_NAME "$REPO_NAME"
+# Adding this value as versioninfo.version ensures we have the same value as would normally
+# show up in a non-hubspot hbase build. Otherwise due to set-maven-versions we'd end up
+# with 2.6-hubspot-SNAPSHOT which is not very useful as a point of reference.
+# Another option would be to pass in our FULL_BUILD_VERSION but that might cause some funniness
+# with the expectations in VersionInfo.compareVersion().
+write-build-env-var MAVEN_BUILD_ARGS "$MAVEN_BUILD_ARGS -Dversioninfo.version=$HBASE_VERSION"
+
+echo "Building HBase version $HBASE_VERSION"
+echo "Will deploy to nexus with version $SET_VERSION"
+echo "Will create rpm with version $FULL_BUILD_VERSION"
+echo "Will run maven with extra args $MAVEN_BUILD_ARGS"

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

@@ -39,10 +39,10 @@ public interface BackupAdmin extends Closeable {
    * Backup given list of tables fully. This is a synchronous operation. It returns backup id on
    * success or throw exception on failure.
    * @param userRequest BackupRequest instance
-   * @return the backup Id
+   * @return backup info
    */
 
-  String backupTables(final BackupRequest userRequest) throws IOException;
+  BackupInfo backupTables(final BackupRequest userRequest) throws IOException;
 
   /**
    * Restore backup
@@ -77,18 +77,10 @@ public interface BackupAdmin extends Closeable {
   void mergeBackups(String[] backupIds) throws IOException;
 
   /**
-   * Show backup history command
-   * @param n last n backup sessions
-   * @return list of backup info objects
-   * @throws IOException exception
-   */
-  List<BackupInfo> getHistory(int n) throws IOException;
-
-  /**
-   * Show backup history command with filters
-   * @param n last n backup sessions
-   * @param f list of filters
-   * @return list of backup info objects
+   * Retrieve info about the most recent backups.
+   * @param n number of backup infos desired
+   * @param f optional filters, only entries passing the filters will be returned
+   * @return a list of at most n entries, ordered from newest (most recent) to oldest (least recent)
    * @throws IOException exception
    */
   List<BackupInfo> getHistory(int n, BackupInfo.Filter... f) throws IOException;

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

@@ -22,6 +22,8 @@
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_BANDWIDTH_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_DEBUG;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_DEBUG_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_HFILE_LOCATION_RESOLVER;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_HFILE_LOCATION_RESOLVER_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_IGNORECHECKSUM;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_IGNORECHECKSUM_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_KEEP;
@@ -35,6 +37,8 @@
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_SET_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_TABLE;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_TABLE_DESC;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_WAL_LOCATION_RESOLVER;
+import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_WAL_LOCATION_RESOLVER_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_WORKERS;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_WORKERS_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_YARN_QUEUE_NAME;
@@ -159,6 +163,8 @@ protected void addOptions() {
     addOptWithArg(OPTION_PATH, OPTION_PATH_DESC);
     addOptWithArg(OPTION_KEEP, OPTION_KEEP_DESC);
     addOptWithArg(OPTION_YARN_QUEUE_NAME, OPTION_YARN_QUEUE_NAME_DESC);
+    addOptWithArg(OPTION_WAL_LOCATION_RESOLVER, OPTION_WAL_LOCATION_RESOLVER_DESC);
+    addOptWithArg(OPTION_HFILE_LOCATION_RESOLVER, OPTION_HFILE_LOCATION_RESOLVER_DESC);
 
   }
 

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

@@ -23,7 +23,6 @@
 import java.util.Set;
 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;
@@ -52,45 +51,57 @@ public class BackupHFileCleaner extends BaseHFileCleanerDelegate implements Abor
   private boolean stopped = false;
   private boolean aborted = false;
   private Connection connection;
-  // 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;
+  private volatile long previousCleaningCompletionTimestamp = 0;
+  private volatile Set<String> bulkLoadedHFilesNeedingBackup = null;
 
   @Override
-  public Iterable<FileStatus> getDeletableFiles(Iterable<FileStatus> files) {
+  public void preClean() {
     if (stopped) {
-      return Collections.emptyList();
+      return;
     }
-
-    // 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);
+      Set<String> hfileFilenames = new HashSet<>();
       for (BulkLoad bulkLoad : tbl.readBulkloadRows(tablesIncludedInBackups)) {
-        hfileFilenames.add(new Path(bulkLoad.getHfilePath()).getName());
+        String path = bulkLoad.getHfilePath();
+        hfileFilenames.add(path.substring(path.lastIndexOf('/') + 1));
       }
-      LOG.debug("Found {} unique HFile filenames registered as bulk loads.", hfileFilenames.size());
+      bulkLoadedHFilesNeedingBackup = hfileFilenames;
+      LOG.debug("Cached {} unique HFile filenames registered as bulk loads.",
+        hfileFilenames.size());
     } catch (IOException ioe) {
-      LOG.error(
-        "Failed to read registered bulk load references from backup system table, marking all files as non-deletable.",
-        ioe);
+      LOG.error("Failed to read registered bulk load references from backup system table, "
+        + "marking all files as non-deletable.", ioe);
+      bulkLoadedHFilesNeedingBackup = null;
+    }
+  }
+
+  @Override
+  public void postClean() {
+    previousCleaningCompletionTimestamp = EnvironmentEdgeManager.currentTime();
+    bulkLoadedHFilesNeedingBackup = null;
+  }
+
+  @Override
+  public Iterable<FileStatus> getDeletableFiles(Iterable<FileStatus> files) {
+    if (stopped) {
+      return Collections.emptyList();
+    }
+
+    final Set<String> hfilesNeedingBackup = bulkLoadedHFilesNeedingBackup;
+    if (hfilesNeedingBackup == null) {
       return Collections.emptyList();
     }
 
-    secondPrevReadFromBackupTbl = prevReadFromBackupTbl;
-    prevReadFromBackupTbl = EnvironmentEdgeManager.currentTime();
+    final long recentFileThreshold = previousCleaningCompletionTimestamp;
 
     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) {
+      if (file.getModificationTime() > recentFileThreshold) {
         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();
-      if (hfileFilenames.contains(hfile)) {
+      if (hfilesNeedingBackup.contains(hfile)) {
         LOG.debug("Preventing deletion due to bulk load registration in backup system table: {}",
           file.getPath().toString());
         return false;

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

@@ -27,6 +27,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.function.Predicate;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.util.BackupUtils;
@@ -46,13 +47,22 @@ public class BackupInfo implements Comparable<BackupInfo> {
   private static final Logger LOG = LoggerFactory.getLogger(BackupInfo.class);
   private static final int MAX_FAILED_MESSAGE_LENGTH = 1024;
 
-  public interface Filter {
-    /**
-     * Filter interface
-     * @param info backup info
-     * @return true if info passes filter, false otherwise
-     */
-    boolean apply(BackupInfo info);
+  public interface Filter extends Predicate<BackupInfo> {
+    /** Returns true if the BackupInfo passes the filter, false otherwise */
+    @Override
+    boolean test(BackupInfo backupInfo);
+  }
+
+  public static Filter withRoot(String backupRoot) {
+    return info -> info.getBackupRootDir().equals(backupRoot);
+  }
+
+  public static Filter withType(BackupType type) {
+    return info -> info.getType() == type;
+  }
+
+  public static Filter withState(BackupState state) {
+    return info -> info.getState() == state;
   }
 
   /**
@@ -61,8 +71,7 @@ public interface Filter {
   public enum BackupState {
     RUNNING,
     COMPLETE,
-    FAILED,
-    ANY
+    FAILED
   }
 
   /**
@@ -170,6 +179,8 @@ public enum BackupPhase {
    */
   private boolean noChecksumVerify;
 
+  private boolean usePreviousLogRoll = false;
+
   public BackupInfo() {
     backupTableInfoMap = new HashMap<>();
   }
@@ -211,6 +222,14 @@ public boolean getNoChecksumVerify() {
     return noChecksumVerify;
   }
 
+  public void setUsePreviousLogRoll(boolean usePreviousLogRoll) {
+    this.usePreviousLogRoll = usePreviousLogRoll;
+  }
+
+  public boolean getUsePreviousLogRoll() {
+    return usePreviousLogRoll;
+  }
+
   public void setBackupTableInfoMap(Map<TableName, BackupTableInfo> backupTableInfoMap) {
     this.backupTableInfoMap = backupTableInfoMap;
   }

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

@@ -75,6 +75,11 @@ public Builder withYarnPoolName(String name) {
       return this;
     }
 
+    public Builder withUsePreviousLogRoll(boolean usePreviousLogRoll) {
+      request.setUsePreviousLogRoll(usePreviousLogRoll);
+      return this;
+    }
+
     public BackupRequest build() {
       return request;
     }
@@ -89,6 +94,7 @@ public BackupRequest build() {
   private boolean noChecksumVerify = false;
   private String backupSetName;
   private String yarnPoolName;
+  private boolean usePreviousLogRoll = false;
 
   private BackupRequest() {
   }
@@ -163,4 +169,12 @@ public String getYarnPoolName() {
   public void setYarnPoolName(String yarnPoolName) {
     this.yarnPoolName = yarnPoolName;
   }
+
+  private void setUsePreviousLogRoll(boolean usePreviousLogRoll) {
+    this.usePreviousLogRoll = usePreviousLogRoll;
+  }
+
+  public boolean getUsePreviousLogRoll() {
+    return this.usePreviousLogRoll;
+  }
 }

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

@@ -99,6 +99,14 @@ public interface BackupRestoreConstants {
   String OPTION_YARN_QUEUE_NAME_DESC = "Yarn queue name to run backup create command on";
   String OPTION_YARN_QUEUE_NAME_RESTORE_DESC = "Yarn queue name to run backup restore command on";
 
+  String OPTION_WAL_LOCATION_RESOLVER = "wal-location-resolver";
+  String OPTION_WAL_LOCATION_RESOLVER_DESC =
+    "WAL file location resolver class for rack-aware incremental backup";
+
+  String OPTION_HFILE_LOCATION_RESOLVER = "hfile-location-resolver";
+  String OPTION_HFILE_LOCATION_RESOLVER_DESC =
+    "HFile location resolver class for rack-aware bulk loading during incremental backup";
+
   String JOB_NAME_CONF_KEY = "mapreduce.job.name";
 
   String BACKUP_CONFIG_STRING = BackupRestoreConstants.BACKUP_ENABLE_KEY + "=true\n"