HBASE-26720 ExportSnapshot should validate the source snapshot before copying files (#4216)

Signed-off-by: Andrew Purtell <apurtell@apache.org>
Signed-off-by: Xiaolin Ha <haxiaolin@apache.org>

Author: xicm

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/snapshot/ExportSnapshot.java

@@ -141,6 +141,8 @@ static final class Options {
         "Do not verify checksum, use name+length only.");
     static final Option NO_TARGET_VERIFY = new Option(null, "no-target-verify", false,
         "Do not verify the integrity of the exported snapshot.");
+    static final Option NO_SOURCE_VERIFY = new Option(null, "no-source-verify", false,
+      "Do not verify the source of the snapshot.");
     static final Option OVERWRITE = new Option(null, "overwrite", false,
         "Rewrite the snapshot manifest if already exists.");
     static final Option CHUSER = new Option(null, "chuser", true,
@@ -903,6 +905,7 @@ private void setPermissionParallel(final FileSystem outputFs, final short filesM
   }
 
   private boolean verifyTarget = true;
+  private boolean verifySource = true;
   private boolean verifyChecksum = true;
   private String snapshotName = null;
   private String targetName = null;
@@ -934,6 +937,7 @@ protected void processOptions(CommandLine cmd) {
     // And verifyChecksum and verifyTarget with values read from old args in processOldArgs(...).
     verifyChecksum = !cmd.hasOption(Options.NO_CHECKSUM_VERIFY.getLongOpt());
     verifyTarget = !cmd.hasOption(Options.NO_TARGET_VERIFY.getLongOpt());
+    verifySource = !cmd.hasOption(Options.NO_SOURCE_VERIFY.getLongOpt());
   }
 
   /**
@@ -984,6 +988,13 @@ public int doWork() throws IOException {
     LOG.debug("outputFs={}, outputRoot={}, skipTmp={}, initialOutputSnapshotDir={}",
       outputFs, outputRoot.toString(), skipTmp, initialOutputSnapshotDir);
 
+    // Verify snapshot source before copying files
+    if (verifySource) {
+      LOG.info("Verify snapshot source, inputFs={}, inputRoot={}, snapshotDir={}.",
+        inputFs.getUri(), inputRoot, snapshotDir);
+      verifySnapshot(srcConf, inputFs, inputRoot, snapshotDir);
+    }
+
     // Find the necessary directory which need to change owner and group
     Path needSetOwnerDir = SnapshotDescriptionUtils.getSnapshotRootDir(outputRoot);
     if (outputFs.exists(needSetOwnerDir)) {
@@ -1134,6 +1145,7 @@ protected void printUsage() {
     addOption(Options.TARGET_NAME);
     addOption(Options.NO_CHECKSUM_VERIFY);
     addOption(Options.NO_TARGET_VERIFY);
+    addOption(Options.NO_SOURCE_VERIFY);
     addOption(Options.OVERWRITE);
     addOption(Options.CHUSER);
     addOption(Options.CHGROUP);