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.HashMap;
027import java.util.List;
028import java.util.Map;
029import java.util.Queue;
030import java.util.concurrent.ConcurrentLinkedQueue;
031import java.util.concurrent.ExecutionException;
032import java.util.concurrent.Future;
033import java.util.concurrent.ThreadFactory;
034import java.util.concurrent.ThreadPoolExecutor;
035import java.util.concurrent.TimeUnit;
036import java.util.concurrent.atomic.AtomicInteger;
037import java.util.function.Function;
038import java.util.stream.Collectors;
039import java.util.stream.Stream;
040import org.apache.hadoop.conf.Configuration;
041import org.apache.hadoop.fs.FileStatus;
042import org.apache.hadoop.fs.FileSystem;
043import org.apache.hadoop.fs.Path;
044import org.apache.hadoop.fs.PathFilter;
045import org.apache.hadoop.hbase.HConstants;
046import org.apache.hadoop.hbase.client.RegionInfo;
047import org.apache.hadoop.hbase.regionserver.HStoreFile;
048import org.apache.hadoop.hbase.util.Bytes;
049import org.apache.hadoop.hbase.util.CommonFSUtils;
050import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
051import org.apache.hadoop.hbase.util.FSUtils;
052import org.apache.hadoop.hbase.util.HFileArchiveUtil;
053import org.apache.hadoop.hbase.util.Threads;
054import org.apache.hadoop.io.MultipleIOException;
055import org.apache.yetus.audience.InterfaceAudience;
056import org.slf4j.Logger;
057import org.slf4j.LoggerFactory;
058
059import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
060
061/**
062 * Utility class to handle the removal of HFiles (or the respective {@link HStoreFile StoreFiles})
063 * for a HRegion from the {@link FileSystem}. The hfiles will be archived or deleted, depending on
064 * the state of the system.
065 */
066@InterfaceAudience.Private
067public class HFileArchiver {
068  private static final Logger LOG = LoggerFactory.getLogger(HFileArchiver.class);
069  private static final String SEPARATOR = ".";
070
071  /** Number of retries in case of fs operation failure */
072  private static final int DEFAULT_RETRIES_NUMBER = 3;
073
074  private static final Function<File, Path> FUNC_FILE_TO_PATH = new Function<File, Path>() {
075    @Override
076    public Path apply(File file) {
077      return file == null ? null : file.getPath();
078    }
079  };
080
081  private static ThreadPoolExecutor archiveExecutor;
082
083  private HFileArchiver() {
084    // hidden ctor since this is just a util
085  }
086
087  /** Returns True if the Region exits in the filesystem. */
088  public static boolean exists(Configuration conf, FileSystem fs, RegionInfo info)
089    throws IOException {
090    Path rootDir = CommonFSUtils.getRootDir(conf);
091    Path regionDir = FSUtils.getRegionDirFromRootDir(rootDir, info);
092    return fs.exists(regionDir);
093  }
094
095  /**
096   * Cleans up all the files for a HRegion by archiving the HFiles to the archive directory
097   * @param conf the configuration to use
098   * @param fs   the file system object
099   * @param info RegionInfo for region to be deleted
100   */
101  public static void archiveRegion(Configuration conf, FileSystem fs, RegionInfo info)
102    throws IOException {
103    Path rootDir = CommonFSUtils.getRootDir(conf);
104    archiveRegion(conf, fs, rootDir, CommonFSUtils.getTableDir(rootDir, info.getTable()),
105      FSUtils.getRegionDirFromRootDir(rootDir, info));
106  }
107
108  /**
109   * Cleans up all the files for a HRegion by archiving the HFiles to the archive directory
110   * @param conf     the configuration to use
111   * @param fs       the file system object
112   * @param info     RegionInfo for region to be deleted
113   * @param rootDir  {@link Path} to the root directory where hbase files are stored (for building
114   *                 the archive path)
115   * @param tableDir {@link Path} to where the table is being stored (for building the archive path)
116   */
117  public static void archiveRegion(Configuration conf, FileSystem fs, RegionInfo info, Path rootDir,
118    Path tableDir) throws IOException {
119    archiveRegion(conf, fs, rootDir, tableDir, FSUtils.getRegionDirFromRootDir(rootDir, info));
120  }
121
122  /**
123   * Remove an entire region from the table directory via archiving the region's hfiles.
124   * @param fs        {@link FileSystem} from which to remove the region
125   * @param rootdir   {@link Path} to the root directory where hbase files are stored (for building
126   *                  the archive path)
127   * @param tableDir  {@link Path} to where the table is being stored (for building the archive
128   *                  path)
129   * @param regionDir {@link Path} to where a region is being stored (for building the archive path)
130   * @return <tt>true</tt> if the region was successfully deleted. <tt>false</tt> if the filesystem
131   *         operations could not complete.
132   * @throws IOException if the request cannot be completed
133   */
134  public static boolean archiveRegion(Configuration conf, FileSystem fs, Path rootdir,
135    Path tableDir, Path regionDir) throws IOException {
136    // otherwise, we archive the files
137    // make sure we can archive
138    if (tableDir == null || regionDir == null) {
139      LOG.error("No archive directory could be found because tabledir (" + tableDir
140        + ") or regiondir (" + regionDir + "was null. Deleting files instead.");
141      if (regionDir != null) {
142        deleteRegionWithoutArchiving(fs, regionDir);
143      }
144      // we should have archived, but failed to. Doesn't matter if we deleted
145      // the archived files correctly or not.
146      return false;
147    }
148
149    LOG.debug("ARCHIVING {}", regionDir);
150
151    // make sure the regiondir lives under the tabledir
152    Preconditions.checkArgument(regionDir.toString().startsWith(tableDir.toString()));
153    Path regionArchiveDir = HFileArchiveUtil.getRegionArchiveDir(rootdir,
154      CommonFSUtils.getTableName(tableDir), regionDir.getName());
155
156    FileStatusConverter getAsFile = new FileStatusConverter(fs);
157    // otherwise, we attempt to archive the store files
158
159    // build collection of just the store directories to archive
160    Collection<File> toArchive = new ArrayList<>();
161    final PathFilter dirFilter = new FSUtils.DirFilter(fs);
162    PathFilter nonHidden = new PathFilter() {
163      @Override
164      public boolean accept(Path file) {
165        return dirFilter.accept(file) && !file.getName().startsWith(".");
166      }
167    };
168    FileStatus[] storeDirs = CommonFSUtils.listStatus(fs, regionDir, nonHidden);
169    // if there no files, we can just delete the directory and return;
170    if (storeDirs == null) {
171      LOG.debug("Directory {} empty.", regionDir);
172      return deleteRegionWithoutArchiving(fs, regionDir);
173    }
174
175    // convert the files in the region to a File
176    Stream.of(storeDirs).map(getAsFile).forEachOrdered(toArchive::add);
177    LOG.debug("Archiving " + toArchive);
178    List<File> failedArchive = resolveAndArchive(conf, fs, regionArchiveDir, toArchive,
179      EnvironmentEdgeManager.currentTime());
180    if (!failedArchive.isEmpty()) {
181      throw new FailedArchiveException(
182        "Failed to archive/delete all the files for region:" + regionDir.getName() + " into "
183          + regionArchiveDir + ". Something is probably awry on the filesystem.",
184        failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList()));
185    }
186    // if that was successful, then we delete the region
187    return deleteRegionWithoutArchiving(fs, regionDir);
188  }
189
190  /**
191   * Archive the specified regions in parallel.
192   * @param conf          the configuration to use
193   * @param fs            {@link FileSystem} from which to remove the region
194   * @param rootDir       {@link Path} to the root directory where hbase files are stored (for
195   *                      building the archive path)
196   * @param tableDir      {@link Path} to where the table is being stored (for building the archive
197   *                      path)
198   * @param regionDirList {@link Path} to where regions are being stored (for building the archive
199   *                      path)
200   * @throws IOException if the request cannot be completed
201   */
202  public static void archiveRegions(Configuration conf, FileSystem fs, Path rootDir, Path tableDir,
203    List<Path> regionDirList) throws IOException {
204    List<Future<Void>> futures = new ArrayList<>(regionDirList.size());
205    for (Path regionDir : regionDirList) {
206      Future<Void> future = getArchiveExecutor(conf).submit(() -> {
207        archiveRegion(conf, fs, rootDir, tableDir, regionDir);
208        return null;
209      });
210      futures.add(future);
211    }
212    try {
213      for (Future<Void> future : futures) {
214        future.get();
215      }
216    } catch (InterruptedException e) {
217      throw new InterruptedIOException(e.getMessage());
218    } catch (ExecutionException e) {
219      throw new IOException(e.getCause());
220    }
221  }
222
223  private static synchronized ThreadPoolExecutor getArchiveExecutor(final Configuration conf) {
224    if (archiveExecutor == null) {
225      int maxThreads = conf.getInt("hbase.hfilearchiver.thread.pool.max", 8);
226      archiveExecutor = Threads.getBoundedCachedThreadPool(maxThreads, 30L, TimeUnit.SECONDS,
227        getThreadFactory("HFileArchiver"));
228
229      // Shutdown this ThreadPool in a shutdown hook
230      Runtime.getRuntime().addShutdownHook(new Thread(() -> archiveExecutor.shutdown()));
231    }
232    return archiveExecutor;
233  }
234
235  // We need this method instead of Threads.getNamedThreadFactory() to pass some tests.
236  // The difference from Threads.getNamedThreadFactory() is that it doesn't fix ThreadGroup for
237  // new threads. If we use Threads.getNamedThreadFactory(), we will face ThreadGroup related
238  // issues in some tests.
239  private static ThreadFactory getThreadFactory(String archiverName) {
240    return new ThreadFactory() {
241      final AtomicInteger threadNumber = new AtomicInteger(1);
242
243      @Override
244      public Thread newThread(Runnable r) {
245        final String name = archiverName + "-" + threadNumber.getAndIncrement();
246        Thread t = new Thread(r, name);
247        t.setDaemon(true);
248        return t;
249      }
250    };
251  }
252
253  /**
254   * Remove from the specified region the store files of the specified column family, either by
255   * archiving them or outright deletion
256   * @param fs       the filesystem where the store files live
257   * @param conf     {@link Configuration} to examine to determine the archive directory
258   * @param parent   Parent region hosting the store files
259   * @param tableDir {@link Path} to where the table is being stored (for building the archive path)
260   * @param family   the family hosting the store files
261   * @throws IOException if the files could not be correctly disposed.
262   */
263  public static void archiveFamily(FileSystem fs, Configuration conf, RegionInfo parent,
264    Path tableDir, byte[] family) throws IOException {
265    Path familyDir = new Path(tableDir, new Path(parent.getEncodedName(), Bytes.toString(family)));
266    archiveFamilyByFamilyDir(fs, conf, parent, familyDir, family);
267  }
268
269  /**
270   * Removes from the specified region the store files of the specified column family, either by
271   * archiving them or outright deletion
272   * @param fs        the filesystem where the store files live
273   * @param conf      {@link Configuration} to examine to determine the archive directory
274   * @param parent    Parent region hosting the store files
275   * @param familyDir {@link Path} to where the family is being stored
276   * @param family    the family hosting the store files
277   * @throws IOException if the files could not be correctly disposed.
278   */
279  public static void archiveFamilyByFamilyDir(FileSystem fs, Configuration conf, RegionInfo parent,
280    Path familyDir, byte[] family) throws IOException {
281    FileStatus[] storeFiles = CommonFSUtils.listStatus(fs, familyDir);
282    if (storeFiles == null) {
283      LOG.debug("No files to dispose of in {}, family={}", parent.getRegionNameAsString(),
284        Bytes.toString(family));
285      return;
286    }
287
288    FileStatusConverter getAsFile = new FileStatusConverter(fs);
289    Collection<File> toArchive = Stream.of(storeFiles).map(getAsFile).collect(Collectors.toList());
290    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, parent, family);
291
292    // do the actual archive
293    List<File> failedArchive =
294      resolveAndArchive(conf, fs, storeArchiveDir, toArchive, EnvironmentEdgeManager.currentTime());
295    if (!failedArchive.isEmpty()) {
296      throw new FailedArchiveException(
297        "Failed to archive/delete all the files for region:"
298          + Bytes.toString(parent.getRegionName()) + ", family:" + Bytes.toString(family) + " into "
299          + storeArchiveDir + ". Something is probably awry on the filesystem.",
300        failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList()));
301    }
302  }
303
304  /**
305   * Remove the store files, either by archiving them or outright deletion
306   * @param conf           {@link Configuration} to examine to determine the archive directory
307   * @param fs             the filesystem where the store files live
308   * @param regionInfo     {@link RegionInfo} of the region hosting the store files
309   * @param family         the family hosting the store files
310   * @param compactedFiles files to be disposed of. No further reading of these files should be
311   *                       attempted; otherwise likely to cause an {@link IOException}
312   * @throws IOException if the files could not be correctly disposed.
313   */
314  public static void archiveStoreFiles(Configuration conf, FileSystem fs, RegionInfo regionInfo,
315    Path tableDir, byte[] family, Collection<HStoreFile> compactedFiles) throws IOException {
316    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, tableDir, family);
317    archive(conf, fs, regionInfo, family, compactedFiles, storeArchiveDir);
318  }
319
320  /**
321   * Archive recovered edits using existing logic for archiving store files. This is currently only
322   * relevant when <b>hbase.region.archive.recovered.edits</b> is true, as recovered edits shouldn't
323   * be kept after replay. In theory, we could use very same method available for archiving store
324   * files, but supporting WAL dir and store files on different FileSystems added the need for extra
325   * validation of the passed FileSystem instance and the path where the archiving edits should be
326   * placed.
327   * @param conf          {@link Configuration} to determine the archive directory.
328   * @param fs            the filesystem used for storing WAL files.
329   * @param regionInfo    {@link RegionInfo} a pseudo region representation for the archiving logic.
330   * @param family        a pseudo familiy representation for the archiving logic.
331   * @param replayedEdits the recovered edits to be archived.
332   * @throws IOException if files can't be achived due to some internal error.
333   */
334  public static void archiveRecoveredEdits(Configuration conf, FileSystem fs, RegionInfo regionInfo,
335    byte[] family, Collection<HStoreFile> replayedEdits) throws IOException {
336    String workingDir = conf.get(CommonFSUtils.HBASE_WAL_DIR, conf.get(HConstants.HBASE_DIR));
337    // extra sanity checks for the right FS
338    Path path = new Path(workingDir);
339    if (path.isAbsoluteAndSchemeAuthorityNull()) {
340      // no schema specified on wal dir value, so it's on same FS as StoreFiles
341      path = new Path(conf.get(HConstants.HBASE_DIR));
342    }
343    if (path.toUri().getScheme() != null && !path.toUri().getScheme().equals(fs.getScheme())) {
344      throw new IOException(
345        "Wrong file system! Should be " + path.toUri().getScheme() + ", but got " + fs.getScheme());
346    }
347    path = HFileArchiveUtil.getStoreArchivePathForRootDir(path, regionInfo, family);
348    archive(conf, fs, regionInfo, family, replayedEdits, path);
349  }
350
351  private static void archive(Configuration conf, FileSystem fs, RegionInfo regionInfo,
352    byte[] family, Collection<HStoreFile> compactedFiles, Path storeArchiveDir) throws IOException {
353    // sometimes in testing, we don't have rss, so we need to check for that
354    if (fs == null) {
355      LOG.warn(
356        "Passed filesystem is null, so just deleting files without archiving for {}," + "family={}",
357        Bytes.toString(regionInfo.getRegionName()), Bytes.toString(family));
358      deleteStoreFilesWithoutArchiving(compactedFiles);
359      return;
360    }
361
362    StoreToFile getStorePath = new StoreToFile(fs);
363    Collection<File> storeFiles =
364      compactedFiles.stream().map(getStorePath).collect(Collectors.toList());
365    archiveFiles(conf, fs, regionInfo, family, storeFiles, storeArchiveDir);
366  }
367
368  private static void archiveFiles(Configuration conf, FileSystem fs, RegionInfo regionInfo,
369    byte[] family, Collection<File> storeFiles, Path storeArchiveDir) throws IOException {
370    // short circuit if we don't have any files to delete
371    if (storeFiles.isEmpty()) {
372      LOG.debug("No files to dispose of, done!");
373      return;
374    }
375
376    // build the archive path
377    if (regionInfo == null || family == null) {
378      throw new IOException("Need to have a region and a family to archive from.");
379    }
380    // make sure we don't archive if we can't and that the archive dir exists
381    if (!fs.mkdirs(storeArchiveDir)) {
382      throw new IOException("Could not make archive directory (" + storeArchiveDir + ") for store:"
383        + Bytes.toString(family) + ", deleting compacted files instead.");
384    }
385
386    // do the actual archive
387    List<File> failedArchive = resolveAndArchive(conf, fs, storeArchiveDir, storeFiles,
388      EnvironmentEdgeManager.currentTime());
389
390    if (!failedArchive.isEmpty()) {
391      throw new FailedArchiveException(
392        "Failed to archive/delete all the files for region:"
393          + Bytes.toString(regionInfo.getRegionName()) + ", family:" + Bytes.toString(family)
394          + " into " + storeArchiveDir + ". Something is probably awry on the filesystem.",
395        failedArchive.stream().map(FUNC_FILE_TO_PATH).collect(Collectors.toList()));
396    }
397  }
398
399  /**
400   * Archive files using their existing statuses, without constructing or closing {@link HStoreFile}
401   * objects. This avoids fetching source file metadata again just to construct store files or
402   * determine their type. Archive conflict detection may still fetch metadata.
403   * @param conf         {@link Configuration} to examine to determine the archive directory
404   * @param fs           the filesystem where the store files live; must be non-null
405   * @param regionInfo   {@link RegionInfo} of the region hosting the store files
406   * @param family       the family hosting the store files
407   * @param fileStatuses the files to be archived; directories are not supported
408   * @throws IOException if an input is not a file or the files could not be archived
409   */
410  public static void archiveStoreFileStatuses(Configuration conf, FileSystem fs,
411    RegionInfo regionInfo, byte[] family, Collection<FileStatus> fileStatuses) throws IOException {
412    // short circuit if we don't have any files to delete
413    if (fileStatuses.isEmpty()) {
414      LOG.debug("No files to dispose of, done!");
415      return;
416    }
417    if (regionInfo == null || family == null) {
418      throw new IOException("Need to have a region and a family to archive from.");
419    }
420    Collection<File> filesToArchive = new ArrayList<>(fileStatuses.size());
421    for (FileStatus status : fileStatuses) {
422      if (!status.isFile()) {
423        throw new IOException("Not a file: " + status.getPath());
424      }
425      filesToArchive.add(new FileableRegularFile(fs, status.getPath()));
426    }
427    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, family);
428    archiveFiles(conf, fs, regionInfo, family, filesToArchive, storeArchiveDir);
429  }
430
431  /**
432   * Archive the store file
433   * @param fs         the filesystem where the store files live
434   * @param regionInfo region hosting the store files
435   * @param conf       {@link Configuration} to examine to determine the archive directory
436   * @param tableDir   {@link Path} to where the table is being stored (for building the archive
437   *                   path)
438   * @param family     the family hosting the store files
439   * @param storeFile  file to be archived
440   * @throws IOException if the files could not be correctly disposed.
441   */
442  public static void archiveStoreFile(Configuration conf, FileSystem fs, RegionInfo regionInfo,
443    Path tableDir, byte[] family, Path storeFile) throws IOException {
444    Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, regionInfo, tableDir, family);
445    // make sure we don't archive if we can't and that the archive dir exists
446    if (!fs.mkdirs(storeArchiveDir)) {
447      throw new IOException("Could not make archive directory (" + storeArchiveDir + ") for store:"
448        + Bytes.toString(family) + ", deleting compacted files instead.");
449    }
450
451    // do the actual archive
452    long start = EnvironmentEdgeManager.currentTime();
453    File file = new FileablePath(fs, storeFile);
454    if (!resolveAndArchiveFile(storeArchiveDir, file, Long.toString(start))) {
455      throw new IOException("Failed to archive/delete the file for region:"
456        + regionInfo.getRegionNameAsString() + ", family:" + Bytes.toString(family) + " into "
457        + storeArchiveDir + ". Something is probably awry on the filesystem.");
458    }
459  }
460
461  /**
462   * Resolve any conflict with an existing archive file via timestamp-append renaming of the
463   * existing file and then archive the passed in files.
464   * @param fs             {@link FileSystem} on which to archive the files
465   * @param baseArchiveDir base archive directory to store the files. If any of the files to archive
466   *                       are directories, will append the name of the directory to the base
467   *                       archive directory name, creating a parallel structure.
468   * @param toArchive      files/directories that need to be archvied
469   * @param start          time the archiving started - used for resolving archive conflicts.
470   * @return the list of failed to archive files.
471   * @throws IOException if an unexpected file operation exception occurred
472   */
473  private static List<File> resolveAndArchive(Configuration conf, FileSystem fs,
474    Path baseArchiveDir, Collection<File> toArchive, long start) throws IOException {
475    // Early exit if no files to archive
476    if (toArchive.isEmpty()) {
477      LOG.trace("No files to archive, returning an empty list.");
478      return Collections.emptyList();
479    }
480
481    LOG.trace("Preparing to archive files into directory: {}", baseArchiveDir);
482
483    // Ensure the archive directory exists
484    ensureArchiveDirectoryExists(fs, baseArchiveDir);
485
486    // Thread-safe collection for storing failures
487    Queue<File> failures = new ConcurrentLinkedQueue<>();
488    String startTime = Long.toString(start);
489
490    // Separate files and directories for processing
491    List<File> filesOnly = new ArrayList<>();
492    for (File file : toArchive) {
493      if (file.isFile()) {
494        filesOnly.add(file);
495      } else {
496        handleDirectory(conf, fs, baseArchiveDir, failures, file, start);
497      }
498    }
499
500    // Archive files concurrently
501    archiveFilesConcurrently(conf, baseArchiveDir, filesOnly, failures, startTime);
502
503    return new ArrayList<>(failures); // Convert to a List for the return value
504  }
505
506  private static void ensureArchiveDirectoryExists(FileSystem fs, Path baseArchiveDir)
507    throws IOException {
508    if (!fs.exists(baseArchiveDir) && !fs.mkdirs(baseArchiveDir)) {
509      throw new IOException("Failed to create the archive directory: " + baseArchiveDir);
510    }
511    LOG.trace("Archive directory ready: {}", baseArchiveDir);
512  }
513
514  private static void handleDirectory(Configuration conf, FileSystem fs, Path baseArchiveDir,
515    Queue<File> failures, File directory, long start) {
516    LOG.trace("Processing directory: {}, archiving its children.", directory);
517    Path subArchiveDir = new Path(baseArchiveDir, directory.getName());
518
519    try {
520      Collection<File> children = directory.getChildren();
521      failures.addAll(resolveAndArchive(conf, fs, subArchiveDir, children, start));
522    } catch (IOException e) {
523      LOG.warn("Failed to archive directory: {}", directory, e);
524      failures.add(directory);
525    }
526  }
527
528  private static void archiveFilesConcurrently(Configuration conf, Path baseArchiveDir,
529    List<File> files, Queue<File> failures, String startTime) {
530    LOG.trace("Archiving {} files concurrently into directory: {}", files.size(), baseArchiveDir);
531    Map<File, Future<Boolean>> futureMap = new HashMap<>();
532    // Submit file archiving tasks
533    // default is 16 which comes equal hbase.hstore.blockingStoreFiles default value
534    int maxThreads = conf.getInt("hbase.hfilearchiver.per.region.thread.pool.max", 16);
535    ThreadPoolExecutor hfilesArchiveExecutor = Threads.getBoundedCachedThreadPool(maxThreads, 30L,
536      TimeUnit.SECONDS, getThreadFactory("HFileArchiverPerRegion-"));
537    try {
538      for (File file : files) {
539        Future<Boolean> future = hfilesArchiveExecutor
540          .submit(() -> resolveAndArchiveFile(baseArchiveDir, file, startTime));
541        futureMap.put(file, future);
542      }
543
544      // Process results of each task
545      for (Map.Entry<File, Future<Boolean>> entry : futureMap.entrySet()) {
546        File file = entry.getKey();
547        try {
548          if (!entry.getValue().get()) {
549            LOG.warn("Failed to archive file: {} into directory: {}", file, baseArchiveDir);
550            failures.add(file);
551          }
552        } catch (InterruptedException e) {
553          LOG.error("Archiving interrupted for file: {}", file, e);
554          Thread.currentThread().interrupt(); // Restore interrupt status
555          failures.add(file);
556        } catch (ExecutionException e) {
557          LOG.error("Archiving failed for file: {}", file, e);
558          failures.add(file);
559        }
560      }
561    } finally {
562      hfilesArchiveExecutor.shutdown();
563    }
564  }
565
566  /**
567   * Attempt to archive the passed in file to the archive directory.
568   * <p>
569   * If the same file already exists in the archive, it is moved to a timestamped directory under
570   * the archive directory and the new file is put in its place.
571   * @param archiveDir       {@link Path} to the directory that stores the archives of the hfiles
572   * @param currentFile      {@link Path} to the original HFile that will be archived
573   * @param archiveStartTime time the archiving started, to resolve naming conflicts
574   * @return <tt>true</tt> if the file is successfully archived. <tt>false</tt> if there was a
575   *         problem, but the operation still completed.
576   * @throws IOException on failure to complete {@link FileSystem} operations.
577   */
578  private static boolean resolveAndArchiveFile(Path archiveDir, File currentFile,
579    String archiveStartTime) throws IOException {
580    // build path as it should be in the archive
581    String filename = currentFile.getName();
582    Path archiveFile = new Path(archiveDir, filename);
583    FileSystem fs = currentFile.getFileSystem();
584
585    // An existing destination file in the archive is unexpected, but we handle it here.
586    if (fs.exists(archiveFile)) {
587      if (!fs.exists(currentFile.getPath())) {
588        // If the file already exists in the archive, and there is no current file to archive, then
589        // assume that the file in archive is correct. This is an unexpected situation, suggesting a
590        // race condition or split brain.
591        // In HBASE-26718 this was found when compaction incorrectly happened during warmupRegion.
592        LOG.warn("{} exists in archive. Attempted to archive nonexistent file {}.", archiveFile,
593          currentFile);
594        // We return success to match existing behavior in this method, where FileNotFoundException
595        // in moveAndClose is ignored.
596        return true;
597      }
598      // There is a conflict between the current file and the already existing archived file.
599      // Move the archived file to a timestamped backup. This is a really, really unlikely
600      // situation, where we get the same name for the existing file, but is included just for that
601      // 1 in trillion chance. We are potentially incurring data loss in the archive directory if
602      // the files are not identical. The timestamped backup will be cleaned by HFileCleaner as it
603      // has no references.
604      FileStatus curStatus = fs.getFileStatus(currentFile.getPath());
605      FileStatus archiveStatus = fs.getFileStatus(archiveFile);
606      long curLen = curStatus.getLen();
607      long archiveLen = archiveStatus.getLen();
608      long curMtime = curStatus.getModificationTime();
609      long archiveMtime = archiveStatus.getModificationTime();
610      if (curLen != archiveLen) {
611        LOG.error(
612          "{} already exists in archive with different size than current {}."
613            + " archiveLen: {} currentLen: {} archiveMtime: {} currentMtime: {}",
614          archiveFile, currentFile, archiveLen, curLen, archiveMtime, curMtime);
615        throw new IOException(
616          archiveFile + " already exists in archive with different size" + " than " + currentFile);
617      }
618
619      LOG.error(
620        "{} already exists in archive, moving to timestamped backup and overwriting"
621          + " current {}. archiveLen: {} currentLen: {} archiveMtime: {} currentMtime: {}",
622        archiveFile, currentFile, archiveLen, curLen, archiveMtime, curMtime);
623
624      // move the archive file to the stamped backup
625      Path backedupArchiveFile = new Path(archiveDir, filename + SEPARATOR + archiveStartTime);
626      if (!fs.rename(archiveFile, backedupArchiveFile)) {
627        LOG.error("Could not rename archive file to backup: " + backedupArchiveFile
628          + ", deleting existing file in favor of newer.");
629        // try to delete the existing file, if we can't rename it
630        if (!fs.delete(archiveFile, false)) {
631          throw new IOException("Couldn't delete existing archive file (" + archiveFile
632            + ") or rename it to the backup file (" + backedupArchiveFile
633            + ") to make room for similarly named file.");
634        }
635      } else {
636        LOG.info("Backed up archive file from {} to {}.", archiveFile, backedupArchiveFile);
637      }
638    }
639
640    LOG.trace("No existing file in archive for {}, free to archive original file.", archiveFile);
641
642    // at this point, we should have a free spot for the archive file
643    boolean success = false;
644    for (int i = 0; !success && i < DEFAULT_RETRIES_NUMBER; ++i) {
645      if (i > 0) {
646        // Ensure that the archive directory exists.
647        // The previous "move to archive" operation has failed probably because
648        // the cleaner has removed our archive directory (HBASE-7643).
649        // (we're in a retry loop, so don't worry too much about the exception)
650        try {
651          if (!fs.exists(archiveDir)) {
652            if (fs.mkdirs(archiveDir)) {
653              LOG.debug("Created archive directory {}", archiveDir);
654            }
655          }
656        } catch (IOException e) {
657          LOG.warn("Failed to create directory {}", archiveDir, e);
658        }
659      }
660
661      try {
662        success = currentFile.moveAndClose(archiveFile);
663      } catch (FileNotFoundException fnfe) {
664        LOG.warn("Failed to archive " + currentFile
665          + " because it does not exist! Skipping and continuing on.", fnfe);
666        success = true;
667      } catch (IOException e) {
668        success = false;
669        // When HFiles are placed on a filesystem other than HDFS a rename operation can be a
670        // non-atomic file copy operation. It can take a long time to copy a large hfile and if
671        // interrupted there may be a partially copied file present at the destination. We must
672        // remove the partially copied file, if any, or otherwise the archive operation will fail
673        // indefinitely from this point.
674        LOG.warn("Failed to archive " + currentFile + " on try #" + i, e);
675        try {
676          fs.delete(archiveFile, false);
677        } catch (FileNotFoundException fnfe) {
678          // This case is fine.
679        } catch (IOException ee) {
680          // Complain about other IO exceptions
681          LOG.warn("Failed to clean up from failure to archive " + currentFile + " on try #" + i,
682            ee);
683        }
684      }
685    }
686
687    if (!success) {
688      LOG.error("Failed to archive " + currentFile);
689      return false;
690    }
691
692    LOG.debug("Archived from {} to {}", currentFile, archiveFile);
693    return true;
694  }
695
696  /**
697   * Without regard for backup, delete a region. Should be used with caution.
698   * @param regionDir {@link Path} to the region to be deleted.
699   * @param fs        FileSystem from which to delete the region
700   * @return <tt>true</tt> on successful deletion, <tt>false</tt> otherwise
701   * @throws IOException on filesystem operation failure
702   */
703  private static boolean deleteRegionWithoutArchiving(FileSystem fs, Path regionDir)
704    throws IOException {
705    if (fs.delete(regionDir, true)) {
706      LOG.debug("Deleted {}", regionDir);
707      return true;
708    }
709    LOG.debug("Failed to delete directory {}", regionDir);
710    return false;
711  }
712
713  /**
714   * Just do a simple delete of the given store files
715   * <p>
716   * A best effort is made to delete each of the files, rather than bailing on the first failure.
717   * <p>
718   * @param compactedFiles store files to delete from the file system.
719   * @throws IOException if a file cannot be deleted. All files will be attempted to deleted before
720   *                     throwing the exception, rather than failing at the first file.
721   */
722  private static void deleteStoreFilesWithoutArchiving(Collection<HStoreFile> compactedFiles)
723    throws IOException {
724    LOG.debug("Deleting files without archiving.");
725    List<IOException> errors = new ArrayList<>(0);
726    for (HStoreFile hsf : compactedFiles) {
727      try {
728        hsf.deleteStoreFile();
729      } catch (IOException e) {
730        LOG.error("Failed to delete {}", hsf.getPath());
731        errors.add(e);
732      }
733    }
734    if (errors.size() > 0) {
735      throw MultipleIOException.createIOException(errors);
736    }
737  }
738
739  /**
740   * Adapt a type to match the {@link File} interface, which is used internally for handling
741   * archival/removal of files
742   * @param <T> type to adapt to the {@link File} interface
743   */
744  private static abstract class FileConverter<T> implements Function<T, File> {
745    protected final FileSystem fs;
746
747    public FileConverter(FileSystem fs) {
748      this.fs = fs;
749    }
750  }
751
752  /**
753   * Convert a FileStatus to something we can manage in the archiving
754   */
755  private static class FileStatusConverter extends FileConverter<FileStatus> {
756    public FileStatusConverter(FileSystem fs) {
757      super(fs);
758    }
759
760    @Override
761    public File apply(FileStatus input) {
762      return new FileablePath(fs, input.getPath());
763    }
764  }
765
766  /**
767   * Convert the {@link HStoreFile} into something we can manage in the archive methods
768   */
769  private static class StoreToFile extends FileConverter<HStoreFile> {
770    public StoreToFile(FileSystem fs) {
771      super(fs);
772    }
773
774    @Override
775    public File apply(HStoreFile input) {
776      return new FileableStoreFile(fs, input);
777    }
778  }
779
780  /**
781   * Wrapper to handle file operations uniformly
782   */
783  private static abstract class File {
784    protected final FileSystem fs;
785
786    public File(FileSystem fs) {
787      this.fs = fs;
788    }
789
790    /**
791     * Delete the file
792     * @throws IOException on failure
793     */
794    abstract void delete() throws IOException;
795
796    /**
797     * Check to see if this is a file or a directory
798     * @return <tt>true</tt> if it is a file, <tt>false</tt> otherwise
799     * @throws IOException on {@link FileSystem} connection error
800     */
801    abstract boolean isFile() throws IOException;
802
803    /**
804     * @return if this is a directory, returns all the children in the directory, otherwise returns
805     *         an empty list
806     */
807    abstract Collection<File> getChildren() throws IOException;
808
809    /**
810     * close any outside readers of the file
811     */
812    abstract void close() throws IOException;
813
814    /** Returns the name of the file (not the full fs path, just the individual file name) */
815    abstract String getName();
816
817    /** Returns the path to this file */
818    abstract Path getPath();
819
820    /**
821     * Move the file to the given destination
822     * @return <tt>true</tt> on success
823     */
824    public boolean moveAndClose(Path dest) throws IOException {
825      this.close();
826      Path p = this.getPath();
827      return CommonFSUtils.renameAndSetModifyTime(fs, p, dest);
828    }
829
830    /** Returns the {@link FileSystem} on which this file resides */
831    public FileSystem getFileSystem() {
832      return this.fs;
833    }
834
835    @Override
836    public String toString() {
837      return this.getClass().getSimpleName() + ", " + getPath().toString();
838    }
839  }
840
841  /**
842   * A {@link File} that wraps a simple {@link Path} on a {@link FileSystem}.
843   */
844  private static class FileablePath extends File {
845    private final Path file;
846
847    public FileablePath(FileSystem fs, Path file) {
848      super(fs);
849      this.file = file;
850    }
851
852    @Override
853    public void delete() throws IOException {
854      if (!fs.delete(file, true)) throw new IOException("Failed to delete:" + this.file);
855    }
856
857    @Override
858    public String getName() {
859      return file.getName();
860    }
861
862    @Override
863    public Collection<File> getChildren() throws IOException {
864      if (fs.isFile(file)) {
865        return Collections.emptyList();
866      }
867      return Stream.of(fs.listStatus(file)).map(new FileStatusConverter(fs))
868        .collect(Collectors.toList());
869    }
870
871    @Override
872    public boolean isFile() throws IOException {
873      return fs.isFile(file);
874    }
875
876    @Override
877    public void close() throws IOException {
878      // NOOP - files are implicitly closed on removal
879    }
880
881    @Override
882    Path getPath() {
883      return file;
884    }
885  }
886
887  /**
888   * {@link File} adapter for a {@link HStoreFile} living on a {@link FileSystem} .
889   */
890  private static class FileableStoreFile extends File {
891    HStoreFile file;
892
893    public FileableStoreFile(FileSystem fs, HStoreFile store) {
894      super(fs);
895      this.file = store;
896    }
897
898    @Override
899    public void delete() throws IOException {
900      file.deleteStoreFile();
901    }
902
903    @Override
904    public String getName() {
905      return file.getPath().getName();
906    }
907
908    @Override
909    public boolean isFile() {
910      return true;
911    }
912
913    @Override
914    public Collection<File> getChildren() throws IOException {
915      // storefiles don't have children
916      return Collections.emptyList();
917    }
918
919    @Override
920    public void close() throws IOException {
921      file.closeStoreFile(true);
922    }
923
924    @Override
925    Path getPath() {
926      return file.getPath();
927    }
928  }
929
930  /**
931   * A path already validated as a regular file by the caller.
932   */
933  private static class FileableRegularFile extends FileablePath {
934    public FileableRegularFile(FileSystem fs, Path path) {
935      super(fs, path);
936    }
937
938    @Override
939    public boolean isFile() {
940      return true;
941    }
942  }
943}