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}