001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.backup;
019
020import java.io.FileNotFoundException;
021import java.io.IOException;
022import java.io.InterruptedIOException;
023import java.util.ArrayList;
024import java.util.Collection;
025import java.util.Collections;
026import java.util.List;
027import java.util.concurrent.ExecutionException;
028import java.util.concurrent.Future;
029import java.util.concurrent.ThreadFactory;
030import java.util.concurrent.ThreadPoolExecutor;
031import java.util.concurrent.TimeUnit;
032import java.util.concurrent.atomic.AtomicInteger;
033import java.util.function.Function;
034import java.util.stream.Collectors;
035import java.util.stream.Stream;
036import org.apache.hadoop.conf.Configuration;
037import org.apache.hadoop.fs.FileStatus;
038import org.apache.hadoop.fs.FileSystem;
039import org.apache.hadoop.fs.Path;
040import org.apache.hadoop.fs.PathFilter;
041import org.apache.hadoop.hbase.HConstants;
042import org.apache.hadoop.hbase.client.RegionInfo;
043import org.apache.hadoop.hbase.regionserver.HStoreFile;
044import org.apache.hadoop.hbase.util.Bytes;
045import org.apache.hadoop.hbase.util.CommonFSUtils;
046import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
047import org.apache.hadoop.hbase.util.FSUtils;
048import org.apache.hadoop.hbase.util.HFileArchiveUtil;
049import org.apache.hadoop.hbase.util.Threads;
050import org.apache.hadoop.io.MultipleIOException;
051import org.apache.yetus.audience.InterfaceAudience;
052import org.slf4j.Logger;
053import org.slf4j.LoggerFactory;
054
055import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
056
057/**
058 * Utility class to handle the removal of HFiles (or the respective {@link HStoreFile StoreFiles})
059 * for a HRegion from the {@link FileSystem}. The hfiles will be archived or deleted, depending on
060 * the state of the system.
061 */
062@InterfaceAudience.Private
063public class HFileArchiver {
064  private static final Logger LOG = LoggerFactory.getLogger(HFileArchiver.class);
065  private static final String SEPARATOR = ".";
066
067  /** Number of retries in case of fs operation failure */
068  private static final int DEFAULT_RETRIES_NUMBER = 3;
069
070  private static final Function<File, Path> FUNC_FILE_TO_PATH =
071      new Function<File, Path>() {
072        @Override
073        public Path apply(File file) {
074          return file == null ? null : file.getPath();
075        }
076      };
077
078  private static ThreadPoolExecutor archiveExecutor;
079
080  private HFileArchiver() {
081    // hidden ctor since this is just a util
082  }
083
084  /**
085   * @return True if the Region exits in the filesystem.
086   */
087  public static boolean exists(Configuration conf, FileSystem fs, RegionInfo info)
088      throws IOException {
089    Path rootDir = FSUtils.getRootDir(conf);
090    Path regionDir = FSUtils.getRegionDirFromRootDir(rootDir, info);
091    return fs.exists(regionDir);
092  }
093
094  /**
095   * Cleans up all the files for a HRegion by archiving the HFiles to the archive directory
096   * @param conf the configuration to use
097   * @param fs the file system object
098   * @param info RegionInfo for region to be deleted
099   */
100  public static void archiveRegion(Configuration conf, FileSystem fs, RegionInfo info)
101      throws IOException {
102    Path rootDir = FSUtils.getRootDir(conf);
103    archiveRegion(fs, rootDir, FSUtils.getTableDir(rootDir, info.getTable()),
104      FSUtils.getRegionDirFromRootDir(rootDir, info));
105  }
106
107  /**
108   * Remove an entire region from the table directory via archiving the region's hfiles.
109   * @param fs {@link FileSystem} from which to remove the region
110   * @param rootdir {@link Path} to the root directory where hbase files are stored (for building
111   *          the archive path)
112   * @param tableDir {@link Path} to where the table is being stored (for building the archive path)
113   * @param regionDir {@link Path} to where a region is being stored (for building the archive path)
114   * @return <tt>true</tt> if the region was successfully deleted. <tt>false</tt> if the filesystem
115   *         operations could not complete.
116   * @throws IOException if the request cannot be completed
117   */
118  public static boolean archiveRegion(FileSystem fs, Path rootdir, Path tableDir, Path regionDir)
119      throws IOException {
120    // otherwise, we archive the files
121    // make sure we can archive
122    if (tableDir == null || regionDir == null) {
123      LOG.error("No archive directory could be found because tabledir (" + tableDir
124          + ") or regiondir (" + regionDir + "was null. Deleting files instead.");
125      if (regionDir != null) {
126        deleteRegionWithoutArchiving(fs, regionDir);
127      }
128      // we should have archived, but failed to. Doesn't matter if we deleted
129      // the archived files correctly or not.
130      return false;
131    }
132
133    LOG.debug("ARCHIVING {}", regionDir);
134
135    // make sure the regiondir lives under the tabledir
136    Preconditions.checkArgument(regionDir.toString().startsWith(tableDir.toString()));
137    Path regionArchiveDir = HFileArchiveUtil.getRegionArchiveDir(rootdir,
138        FSUtils.getTableName(tableDir),
139        regionDir.getName());
140
141    FileStatusConverter getAsFile = new FileStatusConverter(fs);
142    // otherwise, we attempt to archive the store files
143
144    // build collection of just the store directories to archive
145    Collection<File> toArchive = new ArrayList<>();
146    final PathFilter dirFilter = new FSUtils.DirFilter(fs);
147    PathFilter nonHidden = new PathFilter() {
148      @Override
149      public boolean accept(Path file) {
150        return dirFilter.accept(file) && !file.getName().startsWith(".");
151      }
152    };
153    FileStatus[] storeDirs = FSUtils.listStatus(fs, regionDir, nonHidden);
154    // if there no files, we can just delete the directory and return;
155    if (storeDirs == null) {
156      LOG.debug("Directory {} empty.", regionDir);
157      return deleteRegionWithoutArchiving(fs, regionDir);
158    }
159
160    // convert the files in the region to a File
161    Stream.of(storeDirs).map(getAsFile).forEachOrdered(toArchive::add);
162    LOG.debug("Archiving " + toArchive);
163    List<File> failedArchive = resolveAndArchive(fs, regionArchiveDir, toArchive,
164        EnvironmentEdgeManager.currentTime());
165    if (!failedArchive.isEmpty()) {
166      throw new FailedArchiveException(
167        "Failed to archive/delete all the files for region:" + regionDir.getName() + " into " +
168          regionArchiveDir + ". Something is probably awry on the filesystem.",
169        failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList()));
170    }
171    // if that was successful, then we delete the region
172    return deleteRegionWithoutArchiving(fs, regionDir);
173  }
174
175  /**
176   * Archive the specified regions in parallel.
177   * @param conf the configuration to use
178   * @param fs {@link FileSystem} from which to remove the region
179   * @param rootDir {@link Path} to the root directory where hbase files are stored (for building
180   *                            the archive path)
181   * @param tableDir {@link Path} to where the table is being stored (for building the archive
182   *                             path)
183   * @param regionDirList {@link Path} to where regions are being stored (for building the archive
184   *                                  path)
185   * @throws IOException if the request cannot be completed
186   */
187  public static void archiveRegions(Configuration conf, FileSystem fs, Path rootDir, Path tableDir,
188    List<Path> regionDirList) throws IOException {
189    List<Future<Void>> futures = new ArrayList<>(regionDirList.size());
190    for (Path regionDir: regionDirList) {
191      Future<Void> future = getArchiveExecutor(conf).submit(() -> {
192        archiveRegion(fs, rootDir, tableDir, regionDir);
193        return null;
194      });
195      futures.add(future);
196    }
197    try {
198      for (Future<Void> future: futures) {
199        future.get();
200      }
201    } catch (InterruptedException e) {
202      throw new InterruptedIOException(e.getMessage());
203    } catch (ExecutionException e) {
204      throw new IOException(e.getCause());
205    }
206  }
207
208  private static synchronized ThreadPoolExecutor getArchiveExecutor(final Configuration conf) {
209    if (archiveExecutor == null) {
210      int maxThreads = conf.getInt("hbase.hfilearchiver.thread.pool.max", 8);
211      archiveExecutor = Threads.getBoundedCachedThreadPool(maxThreads, 30L, TimeUnit.SECONDS,
212        getThreadFactory());
213
214      // Shutdown this ThreadPool in a shutdown hook
215      Runtime.getRuntime().addShutdownHook(new Thread(() -> archiveExecutor.shutdown()));
216    }
217    return archiveExecutor;
218  }
219
220  // We need this method instead of Threads.getNamedThreadFactory() to pass some tests.
221  // The difference from Threads.getNamedThreadFactory() is that it doesn't fix ThreadGroup for
222  // new threads. If we use Threads.getNamedThreadFactory(), we will face ThreadGroup related
223  // issues in some tests.
224  private static ThreadFactory getThreadFactory() {
225    return new ThreadFactory() {
226      final AtomicInteger threadNumber = new AtomicInteger(1);
227
228      @Override
229      public Thread newThread(Runnable r) {
230        final String name = "HFileArchiver-" + threadNumber.getAndIncrement();
231        Thread t = new Thread(r, name);
232        t.setDaemon(true);
233        return t;
234      }
235    };
236  }
237
238  /**
239   * Remove from the specified region the store files of the specified column family,
240   * either by archiving them or outright deletion
241   * @param fs the filesystem where the store files live
242   * @param conf {@link Configuration} to examine to determine the archive directory
243   * @param parent Parent region hosting the store files
244   * @param tableDir {@link Path} to where the table is being stored (for building the archive path)
245   * @param family the family hosting the store files
246   * @throws IOException if the files could not be correctly disposed.
247   */
248  public static void archiveFamily(FileSystem fs, Configuration conf,
249      RegionInfo parent, Path tableDir, byte[] family) throws IOException {
250    Path familyDir = new Path(tableDir, new Path(parent.getEncodedName(), Bytes.toString(family)));
251    archiveFamilyByFamilyDir(fs, conf, parent, familyDir, family);
252  }
253
254  /**
255   * Removes from the specified region the store files of the specified column family,
256   * either by archiving them or outright deletion
257   * @param fs the filesystem where the store files live
258   * @param conf {@link Configuration} to examine to determine the archive directory
259   * @param parent Parent region hosting the store files
260   * @param familyDir {@link Path} to where the family is being stored
261   * @param family the family hosting the store files
262   * @throws IOException if the files could not be correctly disposed.
263   */
264  public static void archiveFamilyByFamilyDir(FileSystem fs, Configuration conf,
265      RegionInfo parent, Path familyDir, byte[] family) throws IOException {
266    FileStatus[] storeFiles = FSUtils.listStatus(fs, familyDir);
267    if (storeFiles == null) {
268      LOG.debug("No files to dispose of in {}, family={}", parent.getRegionNameAsString(),
269          Bytes.toString(family));
270      return;
271    }
272
273    FileStatusConverter getAsFile = new FileStatusConverter(fs);
274    Collection<File> toArchive = Stream.of(storeFiles).map(getAsFile).collect(Collectors.toList());
275    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, parent, family);
276
277    // do the actual archive
278    List<File> failedArchive = resolveAndArchive(fs, storeArchiveDir, toArchive,
279        EnvironmentEdgeManager.currentTime());
280    if (!failedArchive.isEmpty()){
281      throw new FailedArchiveException("Failed to archive/delete all the files for region:"
282          + Bytes.toString(parent.getRegionName()) + ", family:" + Bytes.toString(family)
283          + " into " + storeArchiveDir + ". Something is probably awry on the filesystem.",
284          failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList()));
285    }
286  }
287
288  /**
289   * Remove the store files, either by archiving them or outright deletion
290   * @param conf {@link Configuration} to examine to determine the archive directory
291   * @param fs the filesystem where the store files live
292   * @param regionInfo {@link RegionInfo} of the region hosting the store files
293   * @param family the family hosting the store files
294   * @param compactedFiles files to be disposed of. No further reading of these files should be
295   *          attempted; otherwise likely to cause an {@link IOException}
296   * @throws IOException if the files could not be correctly disposed.
297   */
298  public static void archiveStoreFiles(Configuration conf, FileSystem fs, RegionInfo regionInfo,
299      Path tableDir, byte[] family, Collection<HStoreFile> compactedFiles)
300      throws IOException {
301    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, tableDir, family);
302    archive(fs, regionInfo, family, compactedFiles, storeArchiveDir);
303  }
304
305  /**
306   * Archive recovered edits using existing logic for archiving store files. This is currently only
307   * relevant when <b>hbase.region.archive.recovered.edits</b> is true, as recovered edits shouldn't
308   * be kept after replay. In theory, we could use very same method available for archiving
309   * store files, but supporting WAL dir and store files on different FileSystems added the need for
310   * extra validation of the passed FileSystem instance and the path where the archiving edits
311   * should be placed.
312   * @param conf {@link Configuration} to determine the archive directory.
313   * @param fs the filesystem used for storing WAL files.
314   * @param regionInfo {@link RegionInfo} a pseudo region representation for the archiving logic.
315   * @param family a pseudo familiy representation for the archiving logic.
316   * @param replayedEdits the recovered edits to be archived.
317   * @throws IOException if files can't be achived due to some internal error.
318   */
319  public static void archiveRecoveredEdits(Configuration conf, FileSystem fs, RegionInfo regionInfo,
320    byte[] family, Collection<HStoreFile> replayedEdits)
321    throws IOException {
322    String workingDir = conf.get(CommonFSUtils.HBASE_WAL_DIR, conf.get(HConstants.HBASE_DIR));
323    //extra sanity checks for the right FS
324    Path path = new Path(workingDir);
325    if(path.isAbsoluteAndSchemeAuthorityNull()){
326      //no schema specified on wal dir value, so it's on same FS as StoreFiles
327      path = new Path(conf.get(HConstants.HBASE_DIR));
328    }
329    if(path.toUri().getScheme()!=null && !path.toUri().getScheme().equals(fs.getScheme())){
330      throw new IOException("Wrong file system! Should be " + path.toUri().getScheme() +
331        ", but got " +  fs.getScheme());
332    }
333    path = HFileArchiveUtil.getStoreArchivePathForRootDir(path, regionInfo, family);
334    archive(fs, regionInfo, family, replayedEdits, path);
335  }
336
337  private static void archive(FileSystem fs, RegionInfo regionInfo, byte[] family,
338    Collection<HStoreFile> compactedFiles, Path storeArchiveDir) throws IOException {
339    // sometimes in testing, we don't have rss, so we need to check for that
340    if (fs == null) {
341      LOG.warn("Passed filesystem is null, so just deleting files without archiving for {}," +
342              "family={}", Bytes.toString(regionInfo.getRegionName()), Bytes.toString(family));
343      deleteStoreFilesWithoutArchiving(compactedFiles);
344      return;
345    }
346
347    // short circuit if we don't have any files to delete
348    if (compactedFiles.isEmpty()) {
349      LOG.debug("No files to dispose of, done!");
350      return;
351    }
352
353    // build the archive path
354    if (regionInfo == null || family == null) throw new IOException(
355        "Need to have a region and a family to archive from.");
356    // make sure we don't archive if we can't and that the archive dir exists
357    if (!fs.mkdirs(storeArchiveDir)) {
358      throw new IOException("Could not make archive directory (" + storeArchiveDir + ") for store:"
359          + Bytes.toString(family) + ", deleting compacted files instead.");
360    }
361
362    // otherwise we attempt to archive the store files
363    LOG.debug("Archiving compacted files.");
364
365    // Wrap the storefile into a File
366    StoreToFile getStorePath = new StoreToFile(fs);
367    Collection<File> storeFiles =
368      compactedFiles.stream().map(getStorePath).collect(Collectors.toList());
369
370    // do the actual archive
371    List<File> failedArchive =
372      resolveAndArchive(fs, storeArchiveDir, storeFiles, EnvironmentEdgeManager.currentTime());
373
374    if (!failedArchive.isEmpty()){
375      throw new FailedArchiveException("Failed to archive/delete all the files for region:"
376          + Bytes.toString(regionInfo.getRegionName()) + ", family:" + Bytes.toString(family)
377          + " into " + storeArchiveDir + ". Something is probably awry on the filesystem.",
378          failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList()));
379    }
380  }
381
382  /**
383   * Archive the store file
384   * @param fs the filesystem where the store files live
385   * @param regionInfo region hosting the store files
386   * @param conf {@link Configuration} to examine to determine the archive directory
387   * @param tableDir {@link Path} to where the table is being stored (for building the archive path)
388   * @param family the family hosting the store files
389   * @param storeFile file to be archived
390   * @throws IOException if the files could not be correctly disposed.
391   */
392  public static void archiveStoreFile(Configuration conf, FileSystem fs, RegionInfo regionInfo,
393      Path tableDir, byte[] family, Path storeFile) throws IOException {
394    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, tableDir, family);
395    // make sure we don't archive if we can't and that the archive dir exists
396    if (!fs.mkdirs(storeArchiveDir)) {
397      throw new IOException("Could not make archive directory (" + storeArchiveDir + ") for store:"
398          + Bytes.toString(family) + ", deleting compacted files instead.");
399    }
400
401    // do the actual archive
402    long start = EnvironmentEdgeManager.currentTime();
403    File file = new FileablePath(fs, storeFile);
404    if (!resolveAndArchiveFile(storeArchiveDir, file, Long.toString(start))) {
405      throw new IOException("Failed to archive/delete the file for region:"
406          + regionInfo.getRegionNameAsString() + ", family:" + Bytes.toString(family)
407          + " into " + storeArchiveDir + ". Something is probably awry on the filesystem.");
408    }
409  }
410
411  /**
412   * Resolve any conflict with an existing archive file via timestamp-append
413   * renaming of the existing file and then archive the passed in files.
414   * @param fs {@link FileSystem} on which to archive the files
415   * @param baseArchiveDir base archive directory to store the files. If any of
416   *          the files to archive are directories, will append the name of the
417   *          directory to the base archive directory name, creating a parallel
418   *          structure.
419   * @param toArchive files/directories that need to be archvied
420   * @param start time the archiving started - used for resolving archive
421   *          conflicts.
422   * @return the list of failed to archive files.
423   * @throws IOException if an unexpected file operation exception occurred
424   */
425  private static List<File> resolveAndArchive(FileSystem fs, Path baseArchiveDir,
426      Collection<File> toArchive, long start) throws IOException {
427    // short circuit if no files to move
428    if (toArchive.isEmpty()) {
429      return Collections.emptyList();
430    }
431
432    LOG.trace("Moving files to the archive directory {}", baseArchiveDir);
433
434    // make sure the archive directory exists
435    if (!fs.exists(baseArchiveDir)) {
436      if (!fs.mkdirs(baseArchiveDir)) {
437        throw new IOException("Failed to create the archive directory:" + baseArchiveDir
438            + ", quitting archive attempt.");
439      }
440      LOG.trace("Created archive directory {}", baseArchiveDir);
441    }
442
443    List<File> failures = new ArrayList<>();
444    String startTime = Long.toString(start);
445    for (File file : toArchive) {
446      // if its a file archive it
447      try {
448        LOG.trace("Archiving {}", file);
449        if (file.isFile()) {
450          // attempt to archive the file
451          if (!resolveAndArchiveFile(baseArchiveDir, file, startTime)) {
452            LOG.warn("Couldn't archive " + file + " into backup directory: " + baseArchiveDir);
453            failures.add(file);
454          }
455        } else {
456          // otherwise its a directory and we need to archive all files
457          LOG.trace("{} is a directory, archiving children files", file);
458          // so we add the directory name to the one base archive
459          Path parentArchiveDir = new Path(baseArchiveDir, file.getName());
460          // and then get all the files from that directory and attempt to
461          // archive those too
462          Collection<File> children = file.getChildren();
463          failures.addAll(resolveAndArchive(fs, parentArchiveDir, children, start));
464        }
465      } catch (IOException e) {
466        LOG.warn("Failed to archive {}", file, e);
467        failures.add(file);
468      }
469    }
470    return failures;
471  }
472
473  /**
474   * Attempt to archive the passed in file to the archive directory.
475   * <p>
476   * If the same file already exists in the archive, it is moved to a timestamped directory under
477   * the archive directory and the new file is put in its place.
478   * @param archiveDir {@link Path} to the directory that stores the archives of the hfiles
479   * @param currentFile {@link Path} to the original HFile that will be archived
480   * @param archiveStartTime time the archiving started, to resolve naming conflicts
481   * @return <tt>true</tt> if the file is successfully archived. <tt>false</tt> if there was a
482   *         problem, but the operation still completed.
483   * @throws IOException on failure to complete {@link FileSystem} operations.
484   */
485  private static boolean resolveAndArchiveFile(Path archiveDir, File currentFile,
486      String archiveStartTime) throws IOException {
487    // build path as it should be in the archive
488    String filename = currentFile.getName();
489    Path archiveFile = new Path(archiveDir, filename);
490    FileSystem fs = currentFile.getFileSystem();
491
492    // if the file already exists in the archive, move that one to a timestamped backup. This is a
493    // really, really unlikely situtation, where we get the same name for the existing file, but
494    // is included just for that 1 in trillion chance.
495    if (fs.exists(archiveFile)) {
496      LOG.debug("{} already exists in archive, moving to timestamped backup and " +
497          "overwriting current.", archiveFile);
498
499      // move the archive file to the stamped backup
500      Path backedupArchiveFile = new Path(archiveDir, filename + SEPARATOR + archiveStartTime);
501      if (!fs.rename(archiveFile, backedupArchiveFile)) {
502        LOG.error("Could not rename archive file to backup: " + backedupArchiveFile
503            + ", deleting existing file in favor of newer.");
504        // try to delete the exisiting file, if we can't rename it
505        if (!fs.delete(archiveFile, false)) {
506          throw new IOException("Couldn't delete existing archive file (" + archiveFile
507              + ") or rename it to the backup file (" + backedupArchiveFile
508              + ") to make room for similarly named file.");
509        }
510      }
511      LOG.debug("Backed up archive file from " + archiveFile);
512    }
513
514    LOG.trace("No existing file in archive for {}, free to archive original file.", archiveFile);
515
516    // at this point, we should have a free spot for the archive file
517    boolean success = false;
518    for (int i = 0; !success && i < DEFAULT_RETRIES_NUMBER; ++i) {
519      if (i > 0) {
520        // Ensure that the archive directory exists.
521        // The previous "move to archive" operation has failed probably because
522        // the cleaner has removed our archive directory (HBASE-7643).
523        // (we're in a retry loop, so don't worry too much about the exception)
524        try {
525          if (!fs.exists(archiveDir)) {
526            if (fs.mkdirs(archiveDir)) {
527              LOG.debug("Created archive directory {}", archiveDir);
528            }
529          }
530        } catch (IOException e) {
531          LOG.warn("Failed to create directory {}", archiveDir, e);
532        }
533      }
534
535      try {
536        success = currentFile.moveAndClose(archiveFile);
537      } catch (FileNotFoundException fnfe) {
538        LOG.warn("Failed to archive " + currentFile +
539            " because it does not exist! Skipping and continuing on.", fnfe);
540        success = true;
541      } catch (IOException e) {
542        LOG.warn("Failed to archive " + currentFile + " on try #" + i, e);
543        success = false;
544      }
545    }
546
547    if (!success) {
548      LOG.error("Failed to archive " + currentFile);
549      return false;
550    }
551
552    LOG.debug("Archived from {} to {}", currentFile, archiveFile);
553    return true;
554  }
555
556  /**
557   * Without regard for backup, delete a region. Should be used with caution.
558   * @param regionDir {@link Path} to the region to be deleted.
559   * @param fs FileSystem from which to delete the region
560   * @return <tt>true</tt> on successful deletion, <tt>false</tt> otherwise
561   * @throws IOException on filesystem operation failure
562   */
563  private static boolean deleteRegionWithoutArchiving(FileSystem fs, Path regionDir)
564      throws IOException {
565    if (fs.delete(regionDir, true)) {
566      LOG.debug("Deleted {}", regionDir);
567      return true;
568    }
569    LOG.debug("Failed to delete directory {}", regionDir);
570    return false;
571  }
572
573  /**
574   * Just do a simple delete of the given store files
575   * <p>
576   * A best effort is made to delete each of the files, rather than bailing on the first failure.
577   * <p>
578   * @param compactedFiles store files to delete from the file system.
579   * @throws IOException if a file cannot be deleted. All files will be attempted to deleted before
580   *           throwing the exception, rather than failing at the first file.
581   */
582  private static void deleteStoreFilesWithoutArchiving(Collection<HStoreFile> compactedFiles)
583      throws IOException {
584    LOG.debug("Deleting files without archiving.");
585    List<IOException> errors = new ArrayList<>(0);
586    for (HStoreFile hsf : compactedFiles) {
587      try {
588        hsf.deleteStoreFile();
589      } catch (IOException e) {
590        LOG.error("Failed to delete {}", hsf.getPath());
591        errors.add(e);
592      }
593    }
594    if (errors.size() > 0) {
595      throw MultipleIOException.createIOException(errors);
596    }
597  }
598
599  /**
600   * Adapt a type to match the {@link File} interface, which is used internally for handling
601   * archival/removal of files
602   * @param <T> type to adapt to the {@link File} interface
603   */
604  private static abstract class FileConverter<T> implements Function<T, File> {
605    protected final FileSystem fs;
606
607    public FileConverter(FileSystem fs) {
608      this.fs = fs;
609    }
610  }
611
612  /**
613   * Convert a FileStatus to something we can manage in the archiving
614   */
615  private static class FileStatusConverter extends FileConverter<FileStatus> {
616    public FileStatusConverter(FileSystem fs) {
617      super(fs);
618    }
619
620    @Override
621    public File apply(FileStatus input) {
622      return new FileablePath(fs, input.getPath());
623    }
624  }
625
626  /**
627   * Convert the {@link HStoreFile} into something we can manage in the archive
628   * methods
629   */
630  private static class StoreToFile extends FileConverter<HStoreFile> {
631    public StoreToFile(FileSystem fs) {
632      super(fs);
633    }
634
635    @Override
636    public File apply(HStoreFile input) {
637      return new FileableStoreFile(fs, input);
638    }
639  }
640
641  /**
642   * Wrapper to handle file operations uniformly
643   */
644  private static abstract class File {
645    protected final FileSystem fs;
646
647    public File(FileSystem fs) {
648      this.fs = fs;
649    }
650
651    /**
652     * Delete the file
653     * @throws IOException on failure
654     */
655    abstract void delete() throws IOException;
656
657    /**
658     * Check to see if this is a file or a directory
659     * @return <tt>true</tt> if it is a file, <tt>false</tt> otherwise
660     * @throws IOException on {@link FileSystem} connection error
661     */
662    abstract boolean isFile() throws IOException;
663
664    /**
665     * @return if this is a directory, returns all the children in the
666     *         directory, otherwise returns an empty list
667     * @throws IOException
668     */
669    abstract Collection<File> getChildren() throws IOException;
670
671    /**
672     * close any outside readers of the file
673     * @throws IOException
674     */
675    abstract void close() throws IOException;
676
677    /**
678     * @return the name of the file (not the full fs path, just the individual
679     *         file name)
680     */
681    abstract String getName();
682
683    /**
684     * @return the path to this file
685     */
686    abstract Path getPath();
687
688    /**
689     * Move the file to the given destination
690     * @param dest
691     * @return <tt>true</tt> on success
692     * @throws IOException
693     */
694    public boolean moveAndClose(Path dest) throws IOException {
695      this.close();
696      Path p = this.getPath();
697      return FSUtils.renameAndSetModifyTime(fs, p, dest);
698    }
699
700    /**
701     * @return the {@link FileSystem} on which this file resides
702     */
703    public FileSystem getFileSystem() {
704      return this.fs;
705    }
706
707    @Override
708    public String toString() {
709      return this.getClass().getSimpleName() + ", " + getPath().toString();
710    }
711  }
712
713  /**
714   * A {@link File} that wraps a simple {@link Path} on a {@link FileSystem}.
715   */
716  private static class FileablePath extends File {
717    private final Path file;
718    private final FileStatusConverter getAsFile;
719
720    public FileablePath(FileSystem fs, Path file) {
721      super(fs);
722      this.file = file;
723      this.getAsFile = new FileStatusConverter(fs);
724    }
725
726    @Override
727    public void delete() throws IOException {
728      if (!fs.delete(file, true)) throw new IOException("Failed to delete:" + this.file);
729    }
730
731    @Override
732    public String getName() {
733      return file.getName();
734    }
735
736    @Override
737    public Collection<File> getChildren() throws IOException {
738      if (fs.isFile(file)) {
739        return Collections.emptyList();
740      }
741      return Stream.of(fs.listStatus(file)).map(getAsFile).collect(Collectors.toList());
742    }
743
744    @Override
745    public boolean isFile() throws IOException {
746      return fs.isFile(file);
747    }
748
749    @Override
750    public void close() throws IOException {
751      // NOOP - files are implicitly closed on removal
752    }
753
754    @Override
755    Path getPath() {
756      return file;
757    }
758  }
759
760  /**
761   * {@link File} adapter for a {@link HStoreFile} living on a {@link FileSystem}
762   * .
763   */
764  private static class FileableStoreFile extends File {
765    HStoreFile file;
766
767    public FileableStoreFile(FileSystem fs, HStoreFile store) {
768      super(fs);
769      this.file = store;
770    }
771
772    @Override
773    public void delete() throws IOException {
774      file.deleteStoreFile();
775    }
776
777    @Override
778    public String getName() {
779      return file.getPath().getName();
780    }
781
782    @Override
783    public boolean isFile() {
784      return true;
785    }
786
787    @Override
788    public Collection<File> getChildren() throws IOException {
789      // storefiles don't have children
790      return Collections.emptyList();
791    }
792
793    @Override
794    public void close() throws IOException {
795      file.closeStoreFile(true);
796    }
797
798    @Override
799    Path getPath() {
800      return file.getPath();
801    }
802  }
803}