Merge branch 'master' into HBASE-29368-key-management-feature

Author: virajjasani

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

@@ -52,10 +52,13 @@ 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;
+  // timestamp of most recent completed cleaning run
+  private volatile long previousCleaningCompletionTimestamp = 0;
+
+  @Override
+  public void postClean() {
+    previousCleaningCompletionTimestamp = EnvironmentEdgeManager.currentTime();
+  }
 
   @Override
   public Iterable<FileStatus> getDeletableFiles(Iterable<FileStatus> files) {
@@ -79,12 +82,12 @@ public Iterable<FileStatus> getDeletableFiles(Iterable<FileStatus> files) {
       return Collections.emptyList();
     }
 
-    secondPrevReadFromBackupTbl = prevReadFromBackupTbl;
-    prevReadFromBackupTbl = EnvironmentEdgeManager.currentTime();
+    // Pin the threshold, we don't want the result to change depending on evaluation time.
+    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;
       }

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

@@ -48,6 +48,7 @@
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
 import org.apache.hadoop.hbase.mapreduce.WALPlayer;
 import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
 import org.apache.hadoop.hbase.snapshot.SnapshotManifest;
@@ -199,6 +200,9 @@ private void mergeSplitAndCopyBulkloadedHFiles(List<String> activeFiles,
         int numActiveFiles = activeFiles.size();
         updateFileLists(activeFiles, archiveFiles);
         if (activeFiles.size() < numActiveFiles) {
+          // We've archived some files, delete bulkloads directory
+          // and re-try
+          deleteBulkLoadDirectory();
           continue;
         }
 
@@ -241,7 +245,7 @@ private void mergeSplitAndCopyBulkloadedHFiles(List<String> files, TableName tn,
     incrementalCopyBulkloadHFiles(tgtFs, tn);
   }
 
-  private void updateFileLists(List<String> activeFiles, List<String> archiveFiles)
+  public void updateFileLists(List<String> activeFiles, List<String> archiveFiles)
     throws IOException {
     List<String> newlyArchived = new ArrayList<>();
 
@@ -251,9 +255,23 @@ private void updateFileLists(List<String> activeFiles, List<String> archiveFiles
       }
     }
 
-    if (newlyArchived.size() > 0) {
+    if (!newlyArchived.isEmpty()) {
+      String rootDir = CommonFSUtils.getRootDir(conf).toString();
+
       activeFiles.removeAll(newlyArchived);
-      archiveFiles.addAll(newlyArchived);
+      for (String file : newlyArchived) {
+        String archivedFile = file.substring(rootDir.length() + 1);
+        Path archivedFilePath = new Path(HFileArchiveUtil.getArchivePath(conf), archivedFile);
+        archivedFile = archivedFilePath.toString();
+
+        if (!fs.exists(archivedFilePath)) {
+          throw new IOException(String.format(
+            "File %s no longer exists, and no archived file %s exists for it", file, archivedFile));
+        }
+
+        LOG.debug("Archived file {} has been updated", archivedFile);
+        archiveFiles.add(archivedFile);
+      }
     }
 
     LOG.debug(newlyArchived.size() + " files have been archived.");
@@ -334,6 +352,7 @@ public void execute() throws IOException, ColumnFamilyMismatchException {
   }
 
   protected void incrementalCopyHFiles(String[] files, String backupDest) throws IOException {
+    boolean diskBasedSortingOriginalValue = HFileOutputFormat2.diskBasedSortingEnabled(conf);
     try {
       LOG.debug("Incremental copy HFiles is starting. dest=" + backupDest);
       // set overall backup phase: incremental_copy
@@ -348,6 +367,7 @@ protected void incrementalCopyHFiles(String[] files, String backupDest) throws I
         LOG.debug("Setting incremental copy HFiles job name to : " + jobname);
       }
       conf.set(JOB_NAME_CONF_KEY, jobname);
+      conf.setBoolean(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY, true);
 
       BackupCopyJob copyService = BackupRestoreFactory.getBackupCopyJob(conf);
       int res = copyService.copy(backupInfo, backupManager, conf, BackupType.INCREMENTAL, strArr);
@@ -360,6 +380,8 @@ protected void incrementalCopyHFiles(String[] files, String backupDest) throws I
         + " finished.");
     } finally {
       deleteBulkLoadDirectory();
+      conf.setBoolean(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY,
+        diskBasedSortingOriginalValue);
     }
   }
 
@@ -413,6 +435,9 @@ protected void walToHFiles(List<String> dirPaths, List<String> tableList) throws
     conf.set(WALPlayer.INPUT_FILES_SEPARATOR_KEY, ";");
     conf.setBoolean(WALPlayer.MULTI_TABLES_SUPPORT, true);
     conf.set(JOB_NAME_CONF_KEY, jobname);
+
+    boolean diskBasedSortingEnabledOriginalValue = HFileOutputFormat2.diskBasedSortingEnabled(conf);
+    conf.setBoolean(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY, true);
     String[] playerArgs = { dirs, StringUtils.join(tableList, ",") };
 
     try {
@@ -421,13 +446,16 @@ protected void walToHFiles(List<String> dirPaths, List<String> tableList) throws
       if (result != 0) {
         throw new IOException("WAL Player failed");
       }
-      conf.unset(WALPlayer.INPUT_FILES_SEPARATOR_KEY);
-      conf.unset(JOB_NAME_CONF_KEY);
     } catch (IOException e) {
       throw e;
     } catch (Exception ee) {
       throw new IOException("Can not convert from directory " + dirs
         + " (check Hadoop, HBase and WALPlayer M/R job logs) ", ee);
+    } finally {
+      conf.setBoolean(HFileOutputFormat2.DISK_BASED_SORTING_ENABLED_KEY,
+        diskBasedSortingEnabledOriginalValue);
+      conf.unset(WALPlayer.INPUT_FILES_SEPARATOR_KEY);
+      conf.unset(JOB_NAME_CONF_KEY);
     }
   }
 

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/mapreduce/MapReduceHFileSplitterJob.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.ExtendedCell;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.PrivateCellUtil;
 import org.apache.hadoop.hbase.TableName;
@@ -34,11 +35,14 @@
 import org.apache.hadoop.hbase.mapreduce.CellSortReducer;
 import org.apache.hadoop.hbase.mapreduce.HFileInputFormat;
 import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
+import org.apache.hadoop.hbase.mapreduce.KeyOnlyCellComparable;
+import org.apache.hadoop.hbase.mapreduce.PreSortedCellsReducer;
 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
 import org.apache.hadoop.hbase.snapshot.SnapshotRegionLocator;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.MapReduceExtendedCell;
 import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
@@ -72,18 +76,28 @@ protected MapReduceHFileSplitterJob(final Configuration c) {
   /**
    * A mapper that just writes out cells. This one can be used together with {@link CellSortReducer}
    */
-  static class HFileCellMapper extends Mapper<NullWritable, Cell, ImmutableBytesWritable, Cell> {
+  static class HFileCellMapper extends Mapper<NullWritable, Cell, WritableComparable<?>, Cell> {
+
+    private boolean diskBasedSortingEnabled = false;
 
     @Override
     public void map(NullWritable key, Cell value, Context context)
       throws IOException, InterruptedException {
-      context.write(new ImmutableBytesWritable(CellUtil.cloneRow(value)),
-        new MapReduceExtendedCell(PrivateCellUtil.ensureExtendedCell(value)));
+      ExtendedCell extendedCell = PrivateCellUtil.ensureExtendedCell(value);
+      context.write(wrap(extendedCell), new MapReduceExtendedCell(extendedCell));
     }
 
     @Override
     public void setup(Context context) throws IOException {
-      // do nothing
+      diskBasedSortingEnabled =
+        HFileOutputFormat2.diskBasedSortingEnabled(context.getConfiguration());
+    }
+
+    private WritableComparable<?> wrap(ExtendedCell cell) {
+      if (diskBasedSortingEnabled) {
+        return new KeyOnlyCellComparable(cell);
+      }
+      return new ImmutableBytesWritable(CellUtil.cloneRow(cell));
     }
   }
 
@@ -107,13 +121,23 @@ public Job createSubmittableJob(String[] args) throws IOException {
       true);
     job.setJarByClass(MapReduceHFileSplitterJob.class);
     job.setInputFormatClass(HFileInputFormat.class);
-    job.setMapOutputKeyClass(ImmutableBytesWritable.class);
     String hfileOutPath = conf.get(BULK_OUTPUT_CONF_KEY);
+    boolean diskBasedSortingEnabled = HFileOutputFormat2.diskBasedSortingEnabled(conf);
+    if (diskBasedSortingEnabled) {
+      job.setMapOutputKeyClass(KeyOnlyCellComparable.class);
+      job.setSortComparatorClass(KeyOnlyCellComparable.KeyOnlyCellComparator.class);
+    } else {
+      job.setMapOutputKeyClass(ImmutableBytesWritable.class);
+    }
     if (hfileOutPath != null) {
       LOG.debug("add incremental job :" + hfileOutPath + " from " + inputDirs);
       TableName tableName = TableName.valueOf(tabName);
       job.setMapperClass(HFileCellMapper.class);
-      job.setReducerClass(CellSortReducer.class);
+      if (diskBasedSortingEnabled) {
+        job.setReducerClass(PreSortedCellsReducer.class);
+      } else {
+        job.setReducerClass(CellSortReducer.class);
+      }
       Path outputDir = new Path(hfileOutPath);
       FileOutputFormat.setOutputPath(job, outputDir);
       job.setMapOutputValueClass(MapReduceExtendedCell.class);

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

@@ -108,11 +108,11 @@ protected Set<TableName> fetchFullyBackedUpTables(BackupSystemTable tbl) {
     Iterable<FileStatus> deletable;
 
     // The first call will not allow any deletions because of the timestamp mechanism.
-    deletable = cleaner.getDeletableFiles(List.of(file1, file1Archived, file2, file3));
+    deletable = callCleaner(cleaner, List.of(file1, file1Archived, file2, file3));
     assertEquals(Set.of(), Sets.newHashSet(deletable));
 
     // No bulk loads registered, so all files can be deleted.
-    deletable = cleaner.getDeletableFiles(List.of(file1, file1Archived, file2, file3));
+    deletable = callCleaner(cleaner, List.of(file1, file1Archived, file2, file3));
     assertEquals(Set.of(file1, file1Archived, file2, file3), Sets.newHashSet(deletable));
 
     // Register some bulk loads.
@@ -125,10 +125,17 @@ protected Set<TableName> fetchFullyBackedUpTables(BackupSystemTable tbl) {
     }
 
     // File 1 can no longer be deleted, because it is registered as a bulk load.
-    deletable = cleaner.getDeletableFiles(List.of(file1, file1Archived, file2, file3));
+    deletable = callCleaner(cleaner, List.of(file1, file1Archived, file2, file3));
     assertEquals(Set.of(file2, file3), Sets.newHashSet(deletable));
   }
 
+  private Iterable<FileStatus> callCleaner(BackupHFileCleaner cleaner, Iterable<FileStatus> files) {
+    cleaner.preClean();
+    Iterable<FileStatus> deletable = cleaner.getDeletableFiles(files);
+    cleaner.postClean();
+    return deletable;
+  }
+
   private FileStatus createFile(String fileName) throws IOException {
     Path file = new Path(root, fileName);
     fs.createNewFile(file);

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

@@ -20,9 +20,11 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import org.apache.hadoop.fs.FileSystem;
@@ -38,6 +40,8 @@
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.tool.BulkLoadHFiles;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.HFileArchiveUtil;
 import org.apache.hadoop.hbase.util.HFileTestUtil;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -147,6 +151,98 @@ private boolean containsRowWithKey(Table table, String rowKey) throws IOExceptio
     return result.containsColumn(famName, qualName);
   }
 
+  @Test
+  public void testUpdateFileListsRaceCondition() throws Exception {
+    try (BackupSystemTable systemTable = new BackupSystemTable(TEST_UTIL.getConnection())) {
+      // Test the race condition where files are archived during incremental backup
+      FileSystem fs = TEST_UTIL.getTestFileSystem();
+
+      String regionName = "region1";
+      String columnFamily = "cf";
+      String filename1 = "hfile1";
+      String filename2 = "hfile2";
+
+      Path rootDir = CommonFSUtils.getRootDir(TEST_UTIL.getConfiguration());
+      Path tableDir = CommonFSUtils.getTableDir(rootDir, table1);
+      Path activeFile1 =
+        new Path(tableDir, regionName + Path.SEPARATOR + columnFamily + Path.SEPARATOR + filename1);
+      Path activeFile2 =
+        new Path(tableDir, regionName + Path.SEPARATOR + columnFamily + Path.SEPARATOR + filename2);
+
+      fs.mkdirs(activeFile1.getParent());
+      fs.create(activeFile1).close();
+      fs.create(activeFile2).close();
+
+      List<String> activeFiles = new ArrayList<>();
+      activeFiles.add(activeFile1.toString());
+      activeFiles.add(activeFile2.toString());
+      List<String> archiveFiles = new ArrayList<>();
+
+      Path archiveDir = HFileArchiveUtil.getStoreArchivePath(TEST_UTIL.getConfiguration(), table1,
+        regionName, columnFamily);
+      Path archivedFile1 = new Path(archiveDir, filename1);
+      fs.mkdirs(archiveDir);
+      assertTrue("File should be moved to archive", fs.rename(activeFile1, archivedFile1));
+
+      TestBackupBase.IncrementalTableBackupClientForTest client =
+        new TestBackupBase.IncrementalTableBackupClientForTest(TEST_UTIL.getConnection(),
+          "test_backup_id",
+          createBackupRequest(BackupType.INCREMENTAL, List.of(table1), BACKUP_ROOT_DIR));
+
+      client.updateFileLists(activeFiles, archiveFiles);
+
+      assertEquals("Only one file should remain in active files", 1, activeFiles.size());
+      assertEquals("File2 should still be in active files", activeFile2.toString(),
+        activeFiles.get(0));
+      assertEquals("One file should be added to archive files", 1, archiveFiles.size());
+      assertEquals("Archived file should have correct path", archivedFile1.toString(),
+        archiveFiles.get(0));
+      systemTable.finishBackupExclusiveOperation();
+    }
+
+  }
+
+  @Test
+  public void testUpdateFileListsMissingArchivedFile() throws Exception {
+    try (BackupSystemTable systemTable = new BackupSystemTable(TEST_UTIL.getConnection())) {
+      // Test that IOException is thrown when file doesn't exist in archive location
+      FileSystem fs = TEST_UTIL.getTestFileSystem();
+
+      String regionName = "region2";
+      String columnFamily = "cf";
+      String filename = "missing_file";
+
+      Path rootDir = CommonFSUtils.getRootDir(TEST_UTIL.getConfiguration());
+      Path tableDir = CommonFSUtils.getTableDir(rootDir, table1);
+      Path activeFile =
+        new Path(tableDir, regionName + Path.SEPARATOR + columnFamily + Path.SEPARATOR + filename);
+
+      fs.mkdirs(activeFile.getParent());
+      fs.create(activeFile).close();
+
+      List<String> activeFiles = new ArrayList<>();
+      activeFiles.add(activeFile.toString());
+      List<String> archiveFiles = new ArrayList<>();
+
+      // Delete the file but don't create it in archive location
+      fs.delete(activeFile, false);
+
+      TestBackupBase.IncrementalTableBackupClientForTest client =
+        new TestBackupBase.IncrementalTableBackupClientForTest(TEST_UTIL.getConnection(),
+          "test_backup_id",
+          createBackupRequest(BackupType.INCREMENTAL, List.of(table1), BACKUP_ROOT_DIR));
+
+      // This should throw IOException since file doesn't exist in archive
+      try {
+        client.updateFileLists(activeFiles, archiveFiles);
+        fail("Expected IOException to be thrown");
+      } catch (IOException e) {
+        // Expected
+      }
+      systemTable.finishBackupExclusiveOperation();
+    }
+  }
+
   private void performBulkLoad(String keyPrefix) throws IOException {
     FileSystem fs = TEST_UTIL.getTestFileSystem();
     Path baseDirectory = TEST_UTIL.getDataTestDirOnTestFS(TEST_NAME);