diff --git a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/MultiTableSnapshotInputFormatImpl.java b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/MultiTableSnapshotInputFormatImpl.java index 42db1db5f87f..cd3785da4db2 100644 --- a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/MultiTableSnapshotInputFormatImpl.java +++ b/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 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()); } } diff --git a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/TableSnapshotInputFormatImpl.java b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/TableSnapshotInputFormatImpl.java index 633a9b25cdd2..324621de4d8c 100644 --- a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/TableSnapshotInputFormatImpl.java +++ b/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()); } diff --git a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java index 36422b6e9f4a..27f9cc8970b6 100644 --- a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java +++ b/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()); } /** diff --git a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/util/MapreduceHFileArchiver.java b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/util/MapreduceHFileArchiver.java new file mode 100644 index 000000000000..80276d95e5db --- /dev/null +++ b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/util/MapreduceHFileArchiver.java @@ -0,0 +1,589 @@ +/* + * 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 java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.PathFilter; +import org.apache.hadoop.hbase.backup.FailedArchiveException; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.snapshot.RestoreSnapshotArchiver; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; + +/** + * MapReduce-local archiver for the snapshot-scanning restore path. It mirrors the server-side + * {@code HFileArchiver} but lives in the MapReduce module so the snapshot-scanning path can be + * evolved independently. It is injected into the shared {@link + * org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper#copySnapshotForScanner} restore/clone + * logic via {@link RestoreSnapshotArchiver}. + */ +@InterfaceAudience.Private +public class MapreduceHFileArchiver implements RestoreSnapshotArchiver { + + private static final Logger LOG = LoggerFactory.getLogger(MapreduceHFileArchiver.class); + private static final String SEPARATOR = "."; + + /** Number of retries in case of fs operation failure */ + private static final int DEFAULT_RETRIES_NUMBER = 3; + + private static final Function FUNC_FILE_TO_PATH = new Function() { + @Override + public Path apply(File file) { + return file == null ? null : file.getPath(); + } + }; + + /** + * Cleans up all the files for a HRegion by archiving the HFiles to the archive directory + * @param conf the configuration to use + * @param fs the file system object + * @param info RegionInfo for region to be deleted + * @param rootDir {@link Path} to the root directory where hbase files are stored (for building + * the archive path) + * @param tableDir {@link Path} to where the table is being stored (for building the archive path) + */ + @Override + public void archiveRegion(Configuration conf, FileSystem fs, RegionInfo info, Path rootDir, + Path tableDir) throws IOException { + archiveRegion(conf, fs, rootDir, tableDir, FSUtils.getRegionDirFromRootDir(rootDir, info)); + } + + /** + * Remove an entire region from the table directory via archiving the region's hfiles. + * @param fs {@link FileSystem} from which to remove the region + * @param rootdir {@link Path} to the root directory where hbase files are stored (for building + * the archive path) + * @param tableDir {@link Path} to where the table is being stored (for building the archive + * path) + * @param regionDir {@link Path} to where a region is being stored (for building the archive path) + * @return true if the region was successfully deleted. false if the filesystem + * operations could not complete. + * @throws IOException if the request cannot be completed + */ + public static boolean archiveRegion(Configuration conf, FileSystem fs, Path rootdir, + Path tableDir, Path regionDir) throws IOException { + // otherwise, we archive the files + // make sure we can archive + if (tableDir == null || regionDir == null) { + LOG.error("No archive directory could be found because tabledir (" + tableDir + + ") or regiondir (" + regionDir + "was null. Deleting files instead."); + if (regionDir != null) { + deleteRegionWithoutArchiving(fs, regionDir); + } + // we should have archived, but failed to. Doesn't matter if we deleted + // the archived files correctly or not. + return false; + } + + LOG.debug("ARCHIVING {}", regionDir); + + // make sure the regiondir lives under the tabledir + Preconditions.checkArgument(regionDir.toString().startsWith(tableDir.toString())); + Path regionArchiveDir = HFileArchiveUtil.getRegionArchiveDir(rootdir, + CommonFSUtils.getTableName(tableDir), regionDir.getName()); + + FileStatusConverter getAsFile = new FileStatusConverter(fs); + // otherwise, we attempt to archive the store files + + // build collection of just the store directories to archive + Collection toArchive = new ArrayList<>(); + final PathFilter dirFilter = new FSUtils.DirFilter(fs); + PathFilter nonHidden = new PathFilter() { + @Override + public boolean accept(Path file) { + return dirFilter.accept(file) && !file.getName().startsWith("."); + } + }; + FileStatus[] storeDirs = CommonFSUtils.listStatus(fs, regionDir, nonHidden); + // if there no files, we can just delete the directory and return; + if (storeDirs == null) { + LOG.debug("Directory {} empty.", regionDir); + return deleteRegionWithoutArchiving(fs, regionDir); + } + + // convert the files in the region to a File + Stream.of(storeDirs).map(getAsFile).forEachOrdered(toArchive::add); + LOG.debug("Archiving " + toArchive); + List failedArchive = resolveAndArchive(conf, fs, regionArchiveDir, toArchive, + EnvironmentEdgeManager.currentTime()); + if (!failedArchive.isEmpty()) { + throw new FailedArchiveException( + "Failed to archive/delete all the files for region:" + regionDir.getName() + " into " + + regionArchiveDir + ". Something is probably awry on the filesystem.", + failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList())); + } + // if that was successful, then we delete the region + return deleteRegionWithoutArchiving(fs, regionDir); + } + + // We need this method instead of Threads.getNamedThreadFactory() to pass some tests. + // The difference from Threads.getNamedThreadFactory() is that it doesn't fix ThreadGroup for + // new threads. If we use Threads.getNamedThreadFactory(), we will face ThreadGroup related + // issues in some tests. + private static ThreadFactory getThreadFactory(String archiverName) { + return new ThreadFactory() { + final AtomicInteger threadNumber = new AtomicInteger(1); + + @Override + public Thread newThread(Runnable r) { + final String name = archiverName + "-" + threadNumber.getAndIncrement(); + Thread t = new Thread(r, name); + t.setDaemon(true); + return t; + } + }; + } + + /** + * Removes from the specified region the store files of the specified column family, either by + * archiving them or outright deletion + * @param fs the filesystem where the store files live + * @param conf {@link Configuration} to examine to determine the archive directory + * @param parent Parent region hosting the store files + * @param familyDir {@link Path} to where the family is being stored + * @param family the family hosting the store files + * @throws IOException if the files could not be correctly disposed. + */ + @Override + public void archiveFamilyByFamilyDir(FileSystem fs, Configuration conf, RegionInfo parent, + Path familyDir, byte[] family) throws IOException { + FileStatus[] storeFiles = CommonFSUtils.listStatus(fs, familyDir); + if (storeFiles == null) { + LOG.debug("No files to dispose of in {}, family={}", parent.getRegionNameAsString(), + Bytes.toString(family)); + return; + } + + FileStatusConverter getAsFile = new FileStatusConverter(fs); + Collection toArchive = Stream.of(storeFiles).map(getAsFile).collect(Collectors.toList()); + Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, parent, family); + + // do the actual archive + List failedArchive = + resolveAndArchive(conf, fs, storeArchiveDir, toArchive, EnvironmentEdgeManager.currentTime()); + if (!failedArchive.isEmpty()) { + throw new FailedArchiveException( + "Failed to archive/delete all the files for region:" + + Bytes.toString(parent.getRegionName()) + ", family:" + Bytes.toString(family) + " into " + + storeArchiveDir + ". Something is probably awry on the filesystem.", + failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList())); + } + } + + /** + * Resolve any conflict with an existing archive file via timestamp-append renaming of the + * existing file and then archive the passed in files. + * @param fs {@link FileSystem} on which to archive the files + * @param baseArchiveDir base archive directory to store the files. If any of the files to archive + * are directories, will append the name of the directory to the base + * archive directory name, creating a parallel structure. + * @param toArchive files/directories that need to be archvied + * @param start time the archiving started - used for resolving archive conflicts. + * @return the list of failed to archive files. + * @throws IOException if an unexpected file operation exception occurred + */ + private static List resolveAndArchive(Configuration conf, FileSystem fs, + Path baseArchiveDir, Collection toArchive, long start) throws IOException { + // Early exit if no files to archive + if (toArchive.isEmpty()) { + LOG.trace("No files to archive, returning an empty list."); + return Collections.emptyList(); + } + + LOG.trace("Preparing to archive files into directory: {}", baseArchiveDir); + + // Ensure the archive directory exists + ensureArchiveDirectoryExists(fs, baseArchiveDir); + + // Thread-safe collection for storing failures + Queue failures = new ConcurrentLinkedQueue<>(); + String startTime = Long.toString(start); + + // Separate files and directories for processing + List filesOnly = new ArrayList<>(); + for (File file : toArchive) { + if (file.isFile()) { + filesOnly.add(file); + } else { + handleDirectory(conf, fs, baseArchiveDir, failures, file, start); + } + } + + // Archive files concurrently + archiveFilesConcurrently(conf, baseArchiveDir, filesOnly, failures, startTime); + + return new ArrayList<>(failures); // Convert to a List for the return value + } + + private static void ensureArchiveDirectoryExists(FileSystem fs, Path baseArchiveDir) + throws IOException { + if (!fs.exists(baseArchiveDir) && !fs.mkdirs(baseArchiveDir)) { + throw new IOException("Failed to create the archive directory: " + baseArchiveDir); + } + LOG.trace("Archive directory ready: {}", baseArchiveDir); + } + + private static void handleDirectory(Configuration conf, FileSystem fs, Path baseArchiveDir, + Queue failures, File directory, long start) { + LOG.trace("Processing directory: {}, archiving its children.", directory); + Path subArchiveDir = new Path(baseArchiveDir, directory.getName()); + + try { + Collection children = directory.getChildren(); + failures.addAll(resolveAndArchive(conf, fs, subArchiveDir, children, start)); + } catch (IOException e) { + LOG.warn("Failed to archive directory: {}", directory, e); + failures.add(directory); + } + } + + private static void archiveFilesConcurrently(Configuration conf, Path baseArchiveDir, + List files, Queue failures, String startTime) { + LOG.trace("Archiving {} files concurrently into directory: {}", files.size(), baseArchiveDir); + Map> futureMap = new HashMap<>(); + // Submit file archiving tasks + // default is 16 which comes equal hbase.hstore.blockingStoreFiles default value + int maxThreads = conf.getInt("hbase.hfilearchiver.per.region.thread.pool.max", 16); + ThreadPoolExecutor hfilesArchiveExecutor = Threads.getBoundedCachedThreadPool(maxThreads, 30L, + TimeUnit.SECONDS, getThreadFactory("HFileArchiverPerRegion-")); + try { + for (File file : files) { + Future future = hfilesArchiveExecutor + .submit(() -> resolveAndArchiveFile(baseArchiveDir, file, startTime)); + futureMap.put(file, future); + } + + // Process results of each task + for (Map.Entry> entry : futureMap.entrySet()) { + File file = entry.getKey(); + try { + if (!entry.getValue().get()) { + LOG.warn("Failed to archive file: {} into directory: {}", file, baseArchiveDir); + failures.add(file); + } + } catch (InterruptedException e) { + LOG.error("Archiving interrupted for file: {}", file, e); + Thread.currentThread().interrupt(); // Restore interrupt status + failures.add(file); + } catch (ExecutionException e) { + LOG.error("Archiving failed for file: {}", file, e); + failures.add(file); + } + } + } finally { + hfilesArchiveExecutor.shutdown(); + } + } + + /** + * Attempt to archive the passed in file to the archive directory. + *

+ * If the same file already exists in the archive, it is moved to a timestamped directory under + * the archive directory and the new file is put in its place. + * @param archiveDir {@link Path} to the directory that stores the archives of the hfiles + * @param currentFile {@link Path} to the original HFile that will be archived + * @param archiveStartTime time the archiving started, to resolve naming conflicts + * @return true if the file is successfully archived. false if there was a + * problem, but the operation still completed. + * @throws IOException on failure to complete {@link FileSystem} operations. + */ + private static boolean resolveAndArchiveFile(Path archiveDir, File currentFile, + String archiveStartTime) throws IOException { + // build path as it should be in the archive + String filename = currentFile.getName(); + Path archiveFile = new Path(archiveDir, filename); + FileSystem fs = currentFile.getFileSystem(); + + // An existing destination file in the archive is unexpected, but we handle it here. + if (fs.exists(archiveFile)) { + if (!fs.exists(currentFile.getPath())) { + // If the file already exists in the archive, and there is no current file to archive, then + // assume that the file in archive is correct. This is an unexpected situation, suggesting a + // race condition or split brain. + // In HBASE-26718 this was found when compaction incorrectly happened during warmupRegion. + LOG.warn("{} exists in archive. Attempted to archive nonexistent file {}.", archiveFile, + currentFile); + // We return success to match existing behavior in this method, where FileNotFoundException + // in moveAndClose is ignored. + return true; + } + // There is a conflict between the current file and the already existing archived file. + // Move the archived file to a timestamped backup. This is a really, really unlikely + // situation, where we get the same name for the existing file, but is included just for that + // 1 in trillion chance. We are potentially incurring data loss in the archive directory if + // the files are not identical. The timestamped backup will be cleaned by HFileCleaner as it + // has no references. + FileStatus curStatus = fs.getFileStatus(currentFile.getPath()); + FileStatus archiveStatus = fs.getFileStatus(archiveFile); + long curLen = curStatus.getLen(); + long archiveLen = archiveStatus.getLen(); + long curMtime = curStatus.getModificationTime(); + long archiveMtime = archiveStatus.getModificationTime(); + if (curLen != archiveLen) { + LOG.error( + "{} already exists in archive with different size than current {}." + + " archiveLen: {} currentLen: {} archiveMtime: {} currentMtime: {}", + archiveFile, currentFile, archiveLen, curLen, archiveMtime, curMtime); + throw new IOException( + archiveFile + " already exists in archive with different size" + " than " + currentFile); + } + + LOG.error( + "{} already exists in archive, moving to timestamped backup and overwriting" + + " current {}. archiveLen: {} currentLen: {} archiveMtime: {} currentMtime: {}", + archiveFile, currentFile, archiveLen, curLen, archiveMtime, curMtime); + + // move the archive file to the stamped backup + Path backedupArchiveFile = new Path(archiveDir, filename + SEPARATOR + archiveStartTime); + if (!fs.rename(archiveFile, backedupArchiveFile)) { + LOG.error("Could not rename archive file to backup: " + backedupArchiveFile + + ", deleting existing file in favor of newer."); + // try to delete the existing file, if we can't rename it + if (!fs.delete(archiveFile, false)) { + throw new IOException("Couldn't delete existing archive file (" + archiveFile + + ") or rename it to the backup file (" + backedupArchiveFile + + ") to make room for similarly named file."); + } + } else { + LOG.info("Backed up archive file from {} to {}.", archiveFile, backedupArchiveFile); + } + } + + LOG.trace("No existing file in archive for {}, free to archive original file.", archiveFile); + + // at this point, we should have a free spot for the archive file + boolean success = false; + for (int i = 0; !success && i < DEFAULT_RETRIES_NUMBER; ++i) { + if (i > 0) { + // Ensure that the archive directory exists. + // The previous "move to archive" operation has failed probably because + // the cleaner has removed our archive directory (HBASE-7643). + // (we're in a retry loop, so don't worry too much about the exception) + try { + if (!fs.exists(archiveDir)) { + if (fs.mkdirs(archiveDir)) { + LOG.debug("Created archive directory {}", archiveDir); + } + } + } catch (IOException e) { + LOG.warn("Failed to create directory {}", archiveDir, e); + } + } + + try { + success = currentFile.moveAndClose(archiveFile); + } catch (FileNotFoundException fnfe) { + LOG.warn("Failed to archive " + currentFile + + " because it does not exist! Skipping and continuing on.", fnfe); + success = true; + } catch (IOException e) { + success = false; + // When HFiles are placed on a filesystem other than HDFS a rename operation can be a + // non-atomic file copy operation. It can take a long time to copy a large hfile and if + // interrupted there may be a partially copied file present at the destination. We must + // remove the partially copied file, if any, or otherwise the archive operation will fail + // indefinitely from this point. + LOG.warn("Failed to archive " + currentFile + " on try #" + i, e); + try { + fs.delete(archiveFile, false); + } catch (FileNotFoundException fnfe) { + // This case is fine. + } catch (IOException ee) { + // Complain about other IO exceptions + LOG.warn("Failed to clean up from failure to archive " + currentFile + " on try #" + i, + ee); + } + } + } + + if (!success) { + LOG.error("Failed to archive " + currentFile); + return false; + } + + LOG.debug("Archived from {} to {}", currentFile, archiveFile); + return true; + } + + /** + * Without regard for backup, delete a region. Should be used with caution. + * @param regionDir {@link Path} to the region to be deleted. + * @param fs FileSystem from which to delete the region + * @return true on successful deletion, false otherwise + * @throws IOException on filesystem operation failure + */ + private static boolean deleteRegionWithoutArchiving(FileSystem fs, Path regionDir) + throws IOException { + if (fs.delete(regionDir, true)) { + LOG.debug("Deleted {}", regionDir); + return true; + } + LOG.debug("Failed to delete directory {}", regionDir); + return false; + } + + + /** + * Adapt a type to match the {@link File} interface, which is used internally for handling + * archival/removal of files + * @param type to adapt to the {@link File} interface + */ + private static abstract class FileConverter implements Function { + protected final FileSystem fs; + + public FileConverter(FileSystem fs) { + this.fs = fs; + } + } + + /** + * Convert a FileStatus to something we can manage in the archiving + */ + private static class FileStatusConverter extends FileConverter { + public FileStatusConverter(FileSystem fs) { + super(fs); + } + + @Override + public File apply(FileStatus input) { + return new FileablePath(fs, input.getPath()); + } + } + + /** + * Wrapper to handle file operations uniformly + */ + private static abstract class File { + protected final FileSystem fs; + + public File(FileSystem fs) { + this.fs = fs; + } + + /** + * Check to see if this is a file or a directory + * @return true if it is a file, false otherwise + * @throws IOException on {@link FileSystem} connection error + */ + abstract boolean isFile() throws IOException; + + /** + * @return if this is a directory, returns all the children in the directory, otherwise returns + * an empty list + */ + abstract Collection getChildren() throws IOException; + + /** + * close any outside readers of the file + */ + abstract void close() throws IOException; + + /** Returns the name of the file (not the full fs path, just the individual file name) */ + abstract String getName(); + + /** Returns the path to this file */ + abstract Path getPath(); + + /** + * Move the file to the given destination + * @return true on success + */ + public boolean moveAndClose(Path dest) throws IOException { + this.close(); + Path p = this.getPath(); + return CommonFSUtils.renameAndSetModifyTime(fs, p, dest); + } + + /** Returns the {@link FileSystem} on which this file resides */ + public FileSystem getFileSystem() { + return this.fs; + } + + @Override + public String toString() { + return this.getClass().getSimpleName() + ", " + getPath().toString(); + } + } + + /** + * A {@link File} that wraps a simple {@link Path} on a {@link FileSystem}. + */ + private static class FileablePath extends File { + private final Path file; + private final FileStatusConverter getAsFile; + + public FileablePath(FileSystem fs, Path file) { + super(fs); + this.file = file; + this.getAsFile = new FileStatusConverter(fs); + } + + @Override + public String getName() { + return file.getName(); + } + + @Override + public Collection getChildren() throws IOException { + if (fs.isFile(file)) { + return Collections.emptyList(); + } + return Stream.of(fs.listStatus(file)).map(getAsFile).collect(Collectors.toList()); + } + + @Override + public boolean isFile() throws IOException { + return fs.isFile(file); + } + + @Override + public void close() throws IOException { + // NOOP - files are implicitly closed on removal + } + + @Override + Path getPath() { + return file; + } + } +} diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestMapreduceSnapshotRestoreGuard.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestMapreduceSnapshotRestoreGuard.java new file mode 100644 index 000000000000..9664f3612aaf --- /dev/null +++ b/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); + } + } +} diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableSnapshotInputFormat.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableSnapshotInputFormat.java index 7a9f8d5dfb63..5d89c50401d5 100644 --- a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableSnapshotInputFormat.java +++ b/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()); diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/util/TestMapreduceHFileArchiver.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/util/TestMapreduceHFileArchiver.java new file mode 100644 index 000000000000..8baad8cd8625 --- /dev/null +++ b/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")); + } + } +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/RestoreSnapshotArchiver.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/RestoreSnapshotArchiver.java new file mode 100644 index 000000000000..7bbca2a5a26d --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/RestoreSnapshotArchiver.java @@ -0,0 +1,66 @@ +/* + * 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.snapshot; + +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.backup.HFileArchiver; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.yetus.audience.InterfaceAudience; + +/** + * Strategy for archiving files removed by {@link RestoreSnapshotHelper} during a restore/clone. + *

+ * The shared restore/clone logic in {@link RestoreSnapshotHelper} only differs between the + * server-side procedures and the MapReduce snapshot-scanning path in how removed files are + * disposed of. Injecting the archiver lets both flows reuse the same core logic instead of forking + * it: server-side callers use {@link #DEFAULT} (backed by {@link HFileArchiver}), while the + * MapReduce module supplies its own implementation. + */ +@InterfaceAudience.Private +public interface RestoreSnapshotArchiver { + + /** + * Archive the hfiles of an entire region that is being removed from the table directory. + */ + void archiveRegion(Configuration conf, FileSystem fs, RegionInfo info, Path rootDir, + Path tableDir) throws IOException; + + /** + * Archive the store files of a single column family that is being removed from a region. + */ + void archiveFamilyByFamilyDir(FileSystem fs, Configuration conf, RegionInfo parent, + Path familyDir, byte[] family) throws IOException; + + /** Default archiver backed by {@link HFileArchiver}, used by server-side restore/clone. */ + RestoreSnapshotArchiver DEFAULT = new RestoreSnapshotArchiver() { + @Override + public void archiveRegion(Configuration conf, FileSystem fs, RegionInfo info, Path rootDir, + Path tableDir) throws IOException { + HFileArchiver.archiveRegion(conf, fs, info, rootDir, tableDir); + } + + @Override + public void archiveFamilyByFamilyDir(FileSystem fs, Configuration conf, RegionInfo parent, + Path familyDir, byte[] family) throws IOException { + HFileArchiver.archiveFamilyByFamilyDir(fs, conf, parent, familyDir, family); + } + }; +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/RestoreSnapshotHelper.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/RestoreSnapshotHelper.java index f0f1ba3899ae..7c7ebe57ecb6 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/RestoreSnapshotHelper.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/snapshot/RestoreSnapshotHelper.java @@ -39,7 +39,6 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.MetaTableAccessor; import org.apache.hadoop.hbase.TableName; -import org.apache.hadoop.hbase.backup.HFileArchiver; import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; import org.apache.hadoop.hbase.client.Connection; @@ -145,6 +144,7 @@ public class RestoreSnapshotHelper { private final Configuration conf; private final FileSystem fs; private final boolean createBackRefs; + private final RestoreSnapshotArchiver archiver; public RestoreSnapshotHelper(final Configuration conf, final FileSystem fs, final SnapshotManifest manifest, final TableDescriptor tableDescriptor, final Path rootDir, @@ -156,6 +156,14 @@ public RestoreSnapshotHelper(final Configuration conf, final FileSystem fs, final SnapshotManifest manifest, final TableDescriptor tableDescriptor, final Path rootDir, final ForeignExceptionDispatcher monitor, final MonitoredTask status, final boolean createBackRefs) { + this(conf, fs, manifest, tableDescriptor, rootDir, monitor, status, createBackRefs, + RestoreSnapshotArchiver.DEFAULT); + } + + public RestoreSnapshotHelper(final Configuration conf, final FileSystem fs, + final SnapshotManifest manifest, final TableDescriptor tableDescriptor, final Path rootDir, + final ForeignExceptionDispatcher monitor, final MonitoredTask status, + final boolean createBackRefs, final RestoreSnapshotArchiver archiver) { this.fs = fs; this.conf = conf; this.snapshotManifest = manifest; @@ -167,6 +175,7 @@ public RestoreSnapshotHelper(final Configuration conf, final FileSystem fs, this.monitor = monitor; this.status = status; this.createBackRefs = createBackRefs; + this.archiver = archiver; } /** @@ -415,7 +424,7 @@ private void removeHdfsRegions(final ThreadPoolExecutor exec, final Listinjected archiver. + */ + @Test + public void testReRestoreInvokesInjectedArchiver() throws IOException, InterruptedException { + TableName tableName = TableName.valueOf("testReRestoreInjectedArchiver"); + Path restoreDir = new Path("/hbase/.tmp-archiver/restore-dest"); + byte[] columnFamily = Bytes.toBytes("A"); + + Table table = TEST_UTIL.createTable(tableName, new byte[][] { columnFamily }, + new byte[][] { new byte[] { 'b' }, new byte[] { 'd' } }); + Put put1 = new Put(Bytes.toBytes("a")); + put1.addColumn(columnFamily, Bytes.toBytes("q"), Bytes.toBytes("val1")); + table.put(put1); + Put put2 = new Put(Bytes.toBytes("b")); + put2.addColumn(columnFamily, Bytes.toBytes("q"), Bytes.toBytes("val2")); + table.put(put2); + Put put3 = new Put(Bytes.toBytes("d")); + put3.addColumn(columnFamily, Bytes.toBytes("q"), Bytes.toBytes("val3")); + table.put(put3); + TEST_UTIL.getAdmin().flush(tableName); + + String snapshotOne = tableName.getNameAsString() + "-snapshot-one"; + createSnapshot(tableName, snapshotOne); + // First restore populates the restore directory with the original (3-region) layout. + RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotOne); + + flipCompactions(false); + mergeRegions(tableName, 2); + String snapshotTwo = tableName.getNameAsString() + "-snapshot-two"; + createSnapshot(tableName, snapshotTwo); + + // Second restore into the SAME directory: the merged-away regions become "regions to remove", + // which RestoreSnapshotHelper archives through the injected archiver. + RecordingArchiver archiver = new RecordingArchiver(); + RestoreSnapshotHelper.copySnapshotForScanner(conf, fs, rootDir, restoreDir, snapshotTwo, + archiver); + flipCompactions(true); + + assertTrue(archiver.regionCount.get() >= 1, + "expected at least one region to be archived via the injected archiver, but got " + + archiver.regionCount.get()); + } + + private void createSnapshot(TableName tableName, String snapshotName) throws IOException { + org.apache.hadoop.hbase.client.SnapshotDescription desc = + new org.apache.hadoop.hbase.client.SnapshotDescription(snapshotName, tableName, + SnapshotType.FLUSH, null, EnvironmentEdgeManager.currentTime(), -1); + TEST_UTIL.getAdmin().snapshot(desc); + assertTrue(TEST_UTIL.getAdmin().listSnapshots().stream() + .anyMatch(ele -> snapshotName.equals(ele.getName()))); + } + + private void flipCompactions(boolean isEnable) { + int numLiveRegionServers = TEST_UTIL.getHBaseCluster().getNumLiveRegionServers(); + for (int serverNumber = 0; serverNumber < numLiveRegionServers; serverNumber++) { + HRegionServer regionServer = TEST_UTIL.getHBaseCluster().getRegionServer(serverNumber); + regionServer.getCompactSplitThread().setCompactionsEnabled(isEnable); + } + } + + private void mergeRegions(TableName tableName, int mergeCount) throws IOException { + List ris = MetaTableAccessor.getTableRegions(TEST_UTIL.getConnection(), tableName); + int originalRegionCount = ris.size(); + assertTrue(originalRegionCount > mergeCount); + RegionInfo[] regionsToMerge = ris.subList(0, mergeCount).toArray(new RegionInfo[] {}); + final ProcedureExecutor procExec = + TEST_UTIL.getHBaseCluster().getMaster().getMasterProcedureExecutor(); + MergeTableRegionsProcedure proc = + new MergeTableRegionsProcedure(procExec.getEnvironment(), regionsToMerge, true); + long procId = procExec.submitProcedure(proc); + ProcedureTestingUtility.waitProcedure(procExec, procId); + ProcedureTestingUtility.assertProcNotFailed(procExec, procId); + assertEquals(originalRegionCount - mergeCount + 1, + MetaTableAccessor.getTableRegions(TEST_UTIL.getConnection(), tableName).size()); + } +}