HBASE-29435 Add MapreduceRestoreSnapshotHelper.java to guard against accidental data loss

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/MultiTableSnapshotInputFormatImpl.java

@@ -32,6 +32,7 @@
 import org.apache.hadoop.hbase.snapshot.SnapshotManifest;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.hadoop.hbase.util.ConfigurationUtil;
+import org.apache.hadoop.hbase.util.MapreduceHFileArchiver;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.yetus.audience.InterfaceStability;
 import org.slf4j.Logger;
@@ -226,7 +227,8 @@ public void restoreSnapshots(Configuration conf, Map<String, Path> snapshotToDir
 
   void restoreSnapshot(Configuration conf, String snapshotName, Path rootDir, Path restoreDir,
     FileSystem fs) throws IOException {
-    RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotName);
+    RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotName,
+      new MapreduceHFileArchiver());
   }
 
 }

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/TableSnapshotInputFormatImpl.java

@@ -49,6 +49,7 @@
 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.MapreduceHFileArchiver;
 import org.apache.hadoop.hbase.util.RegionSplitter;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapreduce.Job;
@@ -616,7 +617,8 @@ public static void setInput(Configuration conf, String snapshotName, Path restor
 
     restoreDir = new Path(restoreDir, UUID.randomUUID().toString());
 
-    RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotName);
+    RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotName,
+      new MapreduceHFileArchiver());
     conf.set(RESTORE_DIR_KEY, restoreDir.toString());
   }
 

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java

@@ -61,6 +61,7 @@
 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.MapreduceHFileArchiver;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hadoop.hbase.util.Strings;
 import org.apache.hadoop.hbase.zookeeper.ZKConfig;
@@ -435,8 +436,10 @@ private void restoreSnapshotForPeerCluster(Configuration conf, String peerQuorum
     FileSystem.setDefaultUri(peerConf, peerFSAddress);
     CommonFSUtils.setRootDir(peerConf, new Path(peerFSAddress, peerHBaseRootAddress));
     FileSystem fs = FileSystem.get(peerConf);
-    RestoreSnapshotHelper.copySnapshotForScanner(peerConf, fs, CommonFSUtils.getRootDir(peerConf),
-      new Path(peerFSAddress, peerSnapshotTmpDir), peerSnapshotName);
+    Path peerRootDir = CommonFSUtils.getRootDir(peerConf);
+    Path peerRestoreDir = new Path(peerFSAddress, peerSnapshotTmpDir);
+    RestoreSnapshotHelper.copySnapshotForScanner(peerConf, fs, peerRootDir, peerRestoreDir,
+      peerSnapshotName, new MapreduceHFileArchiver());
   }
 
   /**

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestMapreduceSnapshotRestoreGuard.java

@@ -0,0 +1,163 @@
+/*
+ * 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.mapreduce;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.master.assignment.AssignmentManager;
+import org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.MapreduceHFileArchiver;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Guard tests (HBASE-29435) for the MapReduce snapshot-scanning restore path. These exercise the
+ * six-argument {@link RestoreSnapshotHelper#copySnapshotForScanner} overload used by the MapReduce
+ * paths, injecting the MapReduce-local {@link MapreduceHFileArchiver}. The guard rejects a restore
+ * directory that would let a MapReduce job archive (and ultimately delete) production HFiles.
+ */
+@Tag(RegionServerTests.TAG)
+@Tag(MediumTests.TAG)
+public class TestMapreduceSnapshotRestoreGuard {
+
+  protected final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+
+  protected Configuration conf;
+  protected FileSystem fs;
+  protected Path rootDir;
+
+  @BeforeAll
+  public static void setupCluster() throws Exception {
+    TEST_UTIL.getConfiguration().setInt(AssignmentManager.ASSIGN_MAX_ATTEMPTS, 3);
+    TEST_UTIL.startMiniCluster();
+  }
+
+  @AfterAll
+  public static void tearDownCluster() throws Exception {
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @BeforeEach
+  public void setup() throws Exception {
+    rootDir = TEST_UTIL.getDataTestDir("testRestore");
+    fs = TEST_UTIL.getTestFileSystem();
+    conf = TEST_UTIL.getConfiguration();
+    CommonFSUtils.setRootDir(conf, rootDir);
+    // Turn off balancer so it doesn't cut in and mess up our placements.
+    TEST_UTIL.getAdmin().balancerSwitch(false, true);
+  }
+
+  /**
+   * Guard (HBASE-29435): a restore directory exactly equal to the HBase root directory must be
+   * rejected before any filesystem work, to avoid archiving/deleting production data.
+   */
+  @Test
+  public void testRejectRestoreDirEqualToRootDir() {
+    Path root = new Path("/hbase");
+    IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+      () -> RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, root, root, "snap",
+        new MapreduceHFileArchiver()));
+    assertTrue(e.getMessage().contains("BLOCKED"), e.getMessage());
+    assertTrue(e.getMessage().contains("cannot be the HBase root directory"), e.getMessage());
+  }
+
+  /**
+   * Guard (HBASE-29435): a restore directory nested under the HBase root directory must be rejected
+   * by the path check (the sub-directory branch).
+   */
+  @Test
+  public void testRejectRestoreDirUnderRootDir() {
+    Path root = new Path("/hbase");
+    Path restoreDir = new Path(root, "data/.tmp-restore");
+    IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+      () -> RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, root, restoreDir, "snap",
+        new MapreduceHFileArchiver()));
+    assertTrue(e.getMessage().contains("BLOCKED"), e.getMessage());
+    assertTrue(e.getMessage().contains("cannot be the HBase root directory or a sub"),
+      e.getMessage());
+  }
+
+  /**
+   * Guard (HBASE-29435): the operation filesystem must host the HBase root directory. Validates the
+   * first filesystem check (the passed-in {@code fs} vs rootDir). A sibling restoreDir is used so
+   * only this check can fire.
+   */
+  @Test
+  public void testRejectMismatchedFilesystem() throws IOException {
+    Path hdfsRoot = TEST_UTIL.getDefaultRootDirPath();
+    FileSystem localFs = FileSystem.getLocal(conf);
+    Path restoreDir = new Path(hdfsRoot.getParent(), "mr-restore-fs");
+    IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+      () -> RestoreSnapshotHelper.copySnapshotForScanner(conf, localFs, hdfsRoot, restoreDir,
+        "snap", new MapreduceHFileArchiver()));
+    assertTrue(e.getMessage().contains("does not match the HBase root directory filesystem"),
+      e.getMessage());
+  }
+
+  /**
+   * Guard (HBASE-29435): the restore directory must live on the same filesystem as the HBase root
+   * directory, even when the operation filesystem already matches root. Validates the second,
+   * distinct filesystem check (restoreDir vs rootDir), which is not covered by the operation-fs
+   * check above.
+   */
+  @Test
+  public void testRejectRestoreDirOnDifferentFilesystem() throws IOException {
+    Path hdfsRoot = TEST_UTIL.getDefaultRootDirPath();
+    FileSystem rootFs = hdfsRoot.getFileSystem(conf);
+    // Operation fs matches root, but restoreDir is explicitly on the local filesystem.
+    Path localRestore = new Path("file:///tmp/mr-restore-" + System.nanoTime());
+    IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+      () -> RestoreSnapshotHelper.copySnapshotForScanner(conf, rootFs, hdfsRoot, localRestore,
+        "snap", new MapreduceHFileArchiver()));
+    assertTrue(e.getMessage().contains("Filesystems for restore directory"), e.getMessage());
+  }
+
+  /**
+   * Negative contract: a restore directory that is a sibling of (not under) the HBase root
+   * directory must not be blocked by the guard. The snapshot does not exist, so any failure comes
+   * from snapshot loading downstream, never from the guard.
+   */
+  @Test
+  public void testSiblingRestoreDirNotBlocked() throws IOException {
+    rootDir = TEST_UTIL.getDefaultRootDirPath();
+    CommonFSUtils.setRootDir(conf, rootDir);
+    fs = rootDir.getFileSystem(conf);
+    Path siblingRestore = new Path("/hbase/.tmp-sibling-restore");
+    try {
+      RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, siblingRestore,
+        "nonexistent-snapshot", new MapreduceHFileArchiver());
+    } catch (Exception e) {
+      String msg = e.getMessage();
+      assertFalse(msg != null && msg.contains("BLOCKED"),
+        "sibling restoreDir must not be blocked by the guard: " + msg);
+    }
+  }
+}

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableSnapshotInputFormat.java

@@ -59,6 +59,7 @@
 import org.apache.hadoop.hbase.testclassification.VerySlowMapReduceTests;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.MapreduceHFileArchiver;
 import org.apache.hadoop.hbase.util.RegionSplitter;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputSplit;
@@ -709,7 +710,7 @@ public void testReadFromRestoredSnapshotViaMR() throws Exception {
         null, snapshotName, rootDir, fs, true);
       Path tempRestoreDir = UTIL.getDataTestDirOnTestFS("restore_" + snapshotName);
       RestoreSnapshotHelper.copySnapshotForScanner(UTIL.getConfiguration(), fs, rootDir,
-        tempRestoreDir, snapshotName);
+        tempRestoreDir, snapshotName, new MapreduceHFileArchiver());
       assertTrue(fs.exists(tempRestoreDir), "Restore directory should exist");
 
       Job job = Job.getInstance(UTIL.getConfiguration());

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/util/TestMapreduceHFileArchiver.java

@@ -0,0 +1,116 @@
+/*
+ * 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.util;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.UUID;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.RegionInfoBuilder;
+import org.apache.hadoop.hbase.snapshot.RestoreSnapshotArchiver;
+import org.apache.hadoop.hbase.testclassification.MapReduceTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Filesystem-level tests for {@link MapreduceHFileArchiver}, the MapReduce-local fork of
+ * {@code HFileArchiver}. They confirm it implements {@link RestoreSnapshotArchiver} and that its
+ * region/family archiving moves store files into the archive directory while removing the source.
+ */
+@Tag(MapReduceTests.TAG)
+@Tag(MediumTests.TAG)
+public class TestMapreduceHFileArchiver {
+
+  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+
+  private Configuration conf;
+  private FileSystem fs;
+  private Path rootDir;
+
+  @BeforeEach
+  public void setup() throws IOException {
+    conf = TEST_UTIL.getConfiguration();
+    rootDir = TEST_UTIL.getDataTestDir("testMrArchiver-" + UUID.randomUUID());
+    fs = rootDir.getFileSystem(conf);
+    CommonFSUtils.setRootDir(conf, rootDir);
+  }
+
+  @AfterEach
+  public void tearDown() throws IOException {
+    fs.delete(rootDir, true);
+  }
+
+  @Test
+  public void testImplementsRestoreSnapshotArchiver() {
+    assertTrue(new MapreduceHFileArchiver() instanceof RestoreSnapshotArchiver);
+  }
+
+  @Test
+  public void testArchiveFamilyByFamilyDir() throws IOException {
+    TableName tableName = TableName.valueOf("testArchiveFamily");
+    RegionInfo region = RegionInfoBuilder.newBuilder(tableName).build();
+    byte[] family = Bytes.toBytes("cf");
+    Path familyDir = new Path(FSUtils.getRegionDirFromRootDir(rootDir, region), "cf");
+    Path storeFile = new Path(familyDir, "f1");
+    writeFile(storeFile);
+    assertTrue(fs.exists(storeFile));
+
+    new MapreduceHFileArchiver().archiveFamilyByFamilyDir(fs, conf, region, familyDir, family);
+
+    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, region, family);
+    assertTrue(fs.exists(new Path(storeArchiveDir, "f1")), "store file should be moved to archive");
+    assertFalse(fs.exists(storeFile), "source store file should be gone after archiving");
+  }
+
+  @Test
+  public void testArchiveRegion() throws IOException {
+    TableName tableName = TableName.valueOf("testArchiveRegion");
+    RegionInfo region = RegionInfoBuilder.newBuilder(tableName).build();
+    Path tableDir = CommonFSUtils.getTableDir(rootDir, tableName);
+    Path regionDir = FSUtils.getRegionDirFromRootDir(rootDir, region);
+    Path storeFile = new Path(new Path(regionDir, "cf"), "f1");
+    writeFile(storeFile);
+    assertTrue(fs.exists(storeFile));
+
+    new MapreduceHFileArchiver().archiveRegion(conf, fs, region, rootDir, tableDir);
+
+    assertFalse(fs.exists(regionDir), "region dir should be removed after archiving");
+    Path regionArchiveDir =
+      HFileArchiveUtil.getRegionArchiveDir(rootDir, tableName, regionDir.getName());
+    assertTrue(fs.exists(new Path(new Path(regionArchiveDir, "cf"), "f1")),
+      "store file should be moved to the region archive");
+  }
+
+  private void writeFile(Path path) throws IOException {
+    fs.mkdirs(path.getParent());
+    try (FSDataOutputStream out = fs.create(path)) {
+      out.write(Bytes.toBytes("data"));
+    }
+  }
+}