Force CF compatibility during incremental backup

Author: Hernan Gelaf-Romer

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

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.backup.impl;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Public
+public final class ColumnFamilyMismatchException extends BackupException {
+  private ColumnFamilyMismatchException(String msg) {
+    super(msg);
+  }
+
+  public static ColumnFamilyMismatchException create(TableName tn,
+    ColumnFamilyDescriptor[] currentCfs, ColumnFamilyDescriptor[] backupCfs) {
+    String currentCfsParsed = StringUtils.join(currentCfs, ',');
+    String backupCfsParsed = StringUtils.join(backupCfs, ',');
+
+    String msg = "Mismatch in column family descriptors for table: " + tn
+      + "\nCurrent families: %s\nBackup families: %s".formatted(currentCfsParsed, backupCfsParsed);
+    return new ColumnFamilyMismatchException(msg);
+  }
+}

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

@@ -23,6 +23,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -32,15 +33,20 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.BackupCopyJob;
+import org.apache.hadoop.hbase.backup.BackupInfo;
 import org.apache.hadoop.hbase.backup.BackupInfo.BackupPhase;
 import org.apache.hadoop.hbase.backup.BackupRequest;
 import org.apache.hadoop.hbase.backup.BackupRestoreFactory;
 import org.apache.hadoop.hbase.backup.BackupType;
+import org.apache.hadoop.hbase.backup.HBackupFileSystem;
 import org.apache.hadoop.hbase.backup.mapreduce.MapReduceBackupCopyJob;
 import org.apache.hadoop.hbase.backup.util.BackupUtils;
 import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.mapreduce.WALPlayer;
+import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
+import org.apache.hadoop.hbase.snapshot.SnapshotManifest;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.hadoop.hbase.util.HFileArchiveUtil;
@@ -51,6 +57,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos;
+
 /**
  * Incremental backup implementation. See the {@link #execute() execute} method.
  */
@@ -262,6 +270,16 @@ private void updateFileLists(List<String> activeFiles, List<String> archiveFiles
   @Override
   public void execute() throws IOException {
     try {
+      Map<TableName, String> tablesToFullBackupIds = getFullBackupIds();
+
+      try (BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+        for (TableName tn : backupInfo.getTables()) {
+          String fullBackupId = tablesToFullBackupIds.get(tn);
+          BackupInfo fullBackupInfo = backupAdmin.getBackupInfo(fullBackupId);
+          verifyHtd(tn, fullBackupInfo);
+        }
+      }
+
       // case PREPARE_INCREMENTAL:
       beginBackup(backupManager, backupInfo);
       backupInfo.setPhase(BackupPhase.PREPARE_INCREMENTAL);
@@ -434,4 +452,63 @@ protected Path getBulkOutputDir() {
     path = new Path(path, backupId);
     return path;
   }
+
+  private Map<TableName, String> getFullBackupIds() throws IOException {
+    // Ancestors are stored from newest to oldest, so we can iterate backwards
+    // in order to populate our backupId map with the most recent full backup
+    // for a given table
+    List<BackupManifest.BackupImage> images = getAncestors(backupInfo);
+    Map<TableName, String> results = new HashMap<>();
+    for (int i = images.size() - 1; i >= 0; i--) {
+      BackupManifest.BackupImage image = images.get(i);
+      if (image.getType() != BackupType.FULL) {
+        continue;
+      }
+
+      for (TableName tn : image.getTableNames()) {
+        results.put(tn, image.getBackupId());
+      }
+    }
+    return results;
+  }
+
+  private void verifyHtd(TableName tn, BackupInfo fullBackupInfo) throws IOException {
+    try (Admin admin = conn.getAdmin()) {
+      ColumnFamilyDescriptor[] currentCfs = admin.getDescriptor(tn).getColumnFamilies();
+      String snapshotName = fullBackupInfo.getSnapshotName(tn);
+      Path root = HBackupFileSystem.getTableBackupPath(tn,
+        new Path(fullBackupInfo.getBackupRootDir()), fullBackupInfo.getBackupId());
+      Path manifestDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(snapshotName, root);
+
+      FileSystem fs;
+      try {
+        fs = FileSystem.get(new URI(fullBackupInfo.getBackupRootDir()), conf);
+      } catch (URISyntaxException e) {
+        throw new IOException("Unable to get fs", e);
+      }
+
+      SnapshotProtos.SnapshotDescription snapshotDescription =
+        SnapshotDescriptionUtils.readSnapshotInfo(fs, manifestDir);
+      SnapshotManifest manifest = SnapshotManifest.open(conf, fs, manifestDir, snapshotDescription);
+
+      ColumnFamilyDescriptor[] backupCfs = manifest.getTableDescriptor().getColumnFamilies();
+      verifyCfs(tn, currentCfs, backupCfs);
+    }
+  }
+
+  private static void verifyCfs(TableName tn, ColumnFamilyDescriptor[] currentCfs,
+    ColumnFamilyDescriptor[] backupCfs) throws IOException {
+    if (currentCfs.length != backupCfs.length) {
+      throw ColumnFamilyMismatchException.create(tn, currentCfs, backupCfs);
+    }
+
+    for (int i = 0; i < backupCfs.length; i++) {
+      String currentCf = currentCfs[i].getNameAsString();
+      String backupCf = backupCfs[i].getNameAsString();
+
+      if (!currentCf.equals(backupCf)) {
+        throw ColumnFamilyMismatchException.create(tn, currentCfs, backupCfs);
+      }
+    }
+  }
 }

hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestIncrementalBackup.java

@@ -18,8 +18,10 @@
 package org.apache.hadoop.hbase.backup;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
@@ -31,6 +33,7 @@
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.impl.BackupAdminImpl;
 import org.apache.hadoop.hbase.backup.impl.BackupManifest;
+import org.apache.hadoop.hbase.backup.impl.ColumnFamilyMismatchException;
 import org.apache.hadoop.hbase.backup.util.BackupUtils;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
@@ -53,6 +56,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.base.Throwables;
 import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
 import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
 
@@ -102,9 +106,7 @@ public void TestIncBackupRestore() throws Exception {
       insertIntoTable(conn, table1, mobName, 3, NB_ROWS_FAM3).close();
       Admin admin = conn.getAdmin();
       BackupAdminImpl client = new BackupAdminImpl(conn);
-      BackupRequest request = createBackupRequest(BackupType.FULL, tables, BACKUP_ROOT_DIR);
-      String backupIdFull = client.backupTables(request);
-      assertTrue(checkSucceeded(backupIdFull));
+      String backupIdFull = takeFullBackup(tables, client);
 
       // #2 - insert some data to table
       Table t1 = insertIntoTable(conn, table1, famName, 1, ADD_ROWS);
@@ -141,16 +143,14 @@ public void TestIncBackupRestore() throws Exception {
         // exception will be thrown.
         LOG.debug("region is not splittable, because " + e);
       }
-      while (!admin.isTableAvailable(table1)) {
-        Thread.sleep(100);
-      }
+      TEST_UTIL.waitTableAvailable(table1);
       long endSplitTime = EnvironmentEdgeManager.currentTime();
       // split finished
       LOG.debug("split finished in =" + (endSplitTime - startSplitTime));
 
       // #3 - incremental backup for multiple tables
       tables = Lists.newArrayList(table1, table2);
-      request = createBackupRequest(BackupType.INCREMENTAL, tables, BACKUP_ROOT_DIR);
+      BackupRequest request = createBackupRequest(BackupType.INCREMENTAL, tables, BACKUP_ROOT_DIR);
       String backupIdIncMultiple = client.backupTables(request);
       assertTrue(checkSucceeded(backupIdIncMultiple));
       BackupManifest manifest =
@@ -165,6 +165,13 @@ public void TestIncBackupRestore() throws Exception {
         .build();
       TEST_UTIL.getAdmin().modifyTable(newTable1Desc);
 
+      // check that an incremental backup fails because the CFs don't match
+      final List<TableName> tablesCopy = tables;
+      IOException ex = assertThrows(IOException.class, () -> client
+        .backupTables(createBackupRequest(BackupType.INCREMENTAL, tablesCopy, BACKUP_ROOT_DIR)));
+      assertEquals(ColumnFamilyMismatchException.class, Throwables.getRootCause(ex).getClass());
+      takeFullBackup(tables, client);
+
       int NB_ROWS_FAM2 = 7;
       Table t3 = insertIntoTable(conn, table1, fam2Name, 2, NB_ROWS_FAM2);
       t3.close();
@@ -227,4 +234,12 @@ public void TestIncBackupRestore() throws Exception {
       admin.close();
     }
   }
+
+  private String takeFullBackup(List<TableName> tables, BackupAdminImpl backupAdmin)
+    throws IOException {
+    BackupRequest req = createBackupRequest(BackupType.FULL, tables, BACKUP_ROOT_DIR);
+    String backupId = backupAdmin.backupTables(req);
+    checkSucceeded(backupId);
+    return backupId;
+  }
 }