HBASE-26271: Cleanup the broken store files under data directory

hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java

@@ -38,7 +38,6 @@
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.PrivateCellUtil;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.regionserver.CellSink;
 import org.apache.hadoop.hbase.regionserver.HMobStore;
 import org.apache.hadoop.hbase.regionserver.HStore;
 import org.apache.hadoop.hbase.regionserver.HStoreFile;
@@ -286,7 +285,6 @@ private void calculateMobLengthMap(SetMultimap<TableName, String> mobRefs) throw
  * </ol>
  * @param fd File details
  * @param scanner Where to read from.
- * @param writer Where to write to.
  * @param smallestReadPoint Smallest read point.
  * @param cleanSeqId When true, remove seqId(used to be mvcc) value which is <= smallestReadPoint
  * @param throughputController The compaction throughput controller.
@@ -295,7 +293,7 @@ private void calculateMobLengthMap(SetMultimap<TableName, String> mobRefs) throw
  * @return Whether compaction ended; false if it was interrupted for any reason.
  */
  @Override
- protected boolean performCompaction(FileDetails fd, InternalScanner scanner, CellSink writer,
+ protected boolean performCompaction(FileDetails fd, InternalScanner scanner,
  long smallestReadPoint, boolean cleanSeqId, ThroughputController throughputController,
  boolean major, int numofFilesToCompact) throws IOException {
  long bytesWrittenProgressForLog = 0;
@@ -665,7 +663,7 @@ private void commitOrAbortMobWriter(StoreFileWriter mobFileWriter, long maxSeqId
 
 
  @Override
- protected List<Path> commitWriter(StoreFileWriter writer, FileDetails fd,
+ protected List<Path> commitWriter(FileDetails fd,
  CompactionRequestImpl request) throws IOException {
  List<Path> newFiles = Lists.newArrayList(writer.getPath());
  writer.appendMetadata(fd.maxSeqId, request.isAllFiles(), request.getFiles());

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMultiFileWriter.java

@@ -110,7 +110,11 @@ public List<Path> abortWriters() {
  return paths;
  }
 
- protected abstract Collection<StoreFileWriter> writers();
+ /**
+ * Returns all writers. This is used to prevent deleting currently writen storefiles
+ * during cleanup.
+ */
+ public abstract Collection<StoreFileWriter> writers();
 
  /**
  * Subclasses override this method to be called at the end of a successful sequence of append; all

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/BrokenStoreFileCleaner.java

@@ -0,0 +1,202 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.ScheduledChore;
+import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.io.HFileLink;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.ipc.RemoteException;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This Chore, every time it runs, will clear the unsused HFiles in the data
+ * folder.
+ */
+@InterfaceAudience.Private
+public class BrokenStoreFileCleaner extends ScheduledChore {
+ private static final Logger LOG = LoggerFactory.getLogger(BrokenStoreFileCleaner.class);
+ public static final String BROKEN_STOREFILE_CLEANER_ENABLED =
+ "hbase.region.broken.storefilecleaner.enabled";
+ public static final boolean DEFAULT_BROKEN_STOREFILE_CLEANER_ENABLED = false;
+ public static final String BROKEN_STOREFILE_CLEANER_TTL =
+ "hbase.region.broken.storefilecleaner.ttl";
+ public static final long DEFAULT_BROKEN_STOREFILE_CLEANER_TTL = 1000 * 60 * 60 * 12; //12h
+ public static final String BROKEN_STOREFILE_CLEANER_DELAY =
+ "hbase.region.broken.storefilecleaner.delay";
+ public static final int DEFAULT_BROKEN_STOREFILE_CLEANER_DELAY = 1000 * 60 * 60 * 2; //2h
+ public static final String BROKEN_STOREFILE_CLEANER_DELAY_JITTER =
+ "hbase.region.broken.storefilecleaner.delay.jitter";
+ public static final double DEFAULT_BROKEN_STOREFILE_CLEANER_DELAY_JITTER = 0.25D;
+ public static final String BROKEN_STOREFILE_CLEANER_PERIOD =
+ "hbase.region.broken.storefilecleaner.period";
+ public static final int DEFAULT_BROKEN_STOREFILE_CLEANER_PERIOD = 1000 * 60 * 60 * 6; //6h
+
+ private HRegionServer regionServer;
+ private final AtomicBoolean enabled = new AtomicBoolean(true);
+ private long fileTtl;
+
+ public BrokenStoreFileCleaner(final int delay, final int period, final Stoppable stopper,
+ Configuration conf, HRegionServer regionServer) {
+ super("BrokenStoreFileCleaner", stopper, period, delay);
+ this.regionServer = regionServer;
+ setEnabled(
+ conf.getBoolean(BROKEN_STOREFILE_CLEANER_ENABLED, DEFAULT_BROKEN_STOREFILE_CLEANER_ENABLED));
+ fileTtl = conf.getLong(BROKEN_STOREFILE_CLEANER_TTL, DEFAULT_BROKEN_STOREFILE_CLEANER_TTL);
+ }
+
+ public boolean setEnabled(final boolean enabled) {
+ return this.enabled.getAndSet(enabled);
+ }
+
+ public boolean getEnabled() {
+ return this.enabled.get();
+ }
+
+ @Override
+ public void chore() {
+ if (getEnabled()) {
+ long start = EnvironmentEdgeManager.currentTime();
+ AtomicLong deletedFiles = new AtomicLong(0);
+ AtomicLong failedDeletes = new AtomicLong(0);
+ for (HRegion region : regionServer.getRegions()) {
+ for (HStore store : region.getStores()) {
+ //only do cleanup in stores not using tmp directories
+ if (store.getStoreEngine().requireWritingToTmpDirFirst()) {
+ continue;
+ }
+ Path storePath =
+ new Path(region.getRegionFileSystem().getRegionDir(), store.getColumnFamilyName());
+
+ try {
+ List<FileStatus> fsStoreFiles =
+ Arrays.asList(region.getRegionFileSystem().fs.listStatus(storePath));
+ fsStoreFiles.forEach(
+ file -> cleanFileIfNeeded(file, store, deletedFiles, failedDeletes));
+ } catch (IOException e) {
+ LOG.warn("Failed to list files in {}, cleanup is skipped there",storePath);
+ continue;
+ }
+ }
+ }
+ LOG.debug(
+ "BrokenStoreFileCleaner on {} run for: {}ms. It deleted {} files and tried but failed "
+ + "to delete {}",
+ regionServer.getServerName().getServerName(), EnvironmentEdgeManager.currentTime() - start,
+ deletedFiles.get(), failedDeletes.get());
+ } else {
+ LOG.trace("Broken storefile Cleaner chore disabled! Not cleaning.");
+ }
+ }
+
+ private void cleanFileIfNeeded(FileStatus file, HStore store,
+ AtomicLong deletedFiles, AtomicLong failedDeletes) {
+ if(file.isDirectory()){
+ LOG.trace("This is a Directory {}, skip cleanup", file.getPath());
+ return;
+ }
+
+ if(!validate(file.getPath())){
+ LOG.trace("Invalid file {}, skip cleanup", file.getPath());
+ return;
+ }
+
+ if(!isOldEnough(file)){
+ LOG.trace("Fresh file {}, skip cleanup", file.getPath());
+ return;
+ }
+
+ if(isActiveStorefile(file, store)){
+ LOG.trace("Actively used storefile file {}, skip cleanup", file.getPath());
+ return;
+ }
+
+ // Compacted files can still have readers and are cleaned by a separate chore, so they have to
+ // be skipped here
+ if(isCompactedFile(file, store)){
+ LOG.trace("Cleanup is done by a different chore for file {}, skip cleanup", file.getPath());
+ return;
+ }
+
+ if(isCompactionResultFile(file, store)){
+ LOG.trace("The file is the result of an ongoing compaction {}, skip cleanup", file.getPath());
+ return;
+ }
+
+ deleteFile(file, store, deletedFiles, failedDeletes);
+ }
+
+ private boolean isCompactionResultFile(FileStatus file, HStore store) {
+ return store.getStoreEngine().getCompactor().getCompactionTargets().contains(file.getPath());
+ }
+
+ // Compacted files can still have readers and are cleaned by a separate chore, so they have to
+ // be skipped here
+ private boolean isCompactedFile(FileStatus file, HStore store) {
+ return store.getStoreEngine().getStoreFileManager().getCompactedfiles().stream()
+ .anyMatch(sf -> sf.getPath().equals(file.getPath()));
+ }
+
+ private boolean isActiveStorefile(FileStatus file, HStore store) {
+ return store.getStoreEngine().getStoreFileManager().getStorefiles().stream()
+ .anyMatch(sf -> sf.getPath().equals(file.getPath()));
+ }
+
+ boolean validate(Path file) {
+ if (HFileLink.isBackReferencesDir(file) || HFileLink.isBackReferencesDir(file.getParent())) {
+ return true;
+ }
+ return StoreFileInfo.validateStoreFileName(file.getName());
+ }
+
+ boolean isOldEnough(FileStatus file){
+ return file.getModificationTime() + fileTtl < EnvironmentEdgeManager.currentTime();
+ }
+
+ private void deleteFile(FileStatus file, HStore store, AtomicLong deletedFiles,
+ AtomicLong failedDeletes) {
+ Path filePath = file.getPath();
+ LOG.debug("Removing {} from store", filePath);
+ try {
+ boolean success = store.getFileSystem().delete(filePath, false);
+ if (!success) {
+ failedDeletes.incrementAndGet();
+ LOG.warn("Attempted to delete:" + filePath
+ + ", but couldn't. Attempt to delete on next pass.");
+ }
+ else{
+ deletedFiles.incrementAndGet();
+ }
+ } catch (IOException e) {
+ e = e instanceof RemoteException ?
+ ((RemoteException)e).unwrapRemoteException() : e;
+ LOG.warn("Error while deleting: " + filePath, e);
+ }
+ }
+
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DateTieredMultiFileWriter.java

@@ -71,7 +71,7 @@ public void append(Cell cell) throws IOException {
  }
 
  @Override
-protected Collection<StoreFileWriter> writers() {
+public Collection<StoreFileWriter> writers() {
  return lowerBoundary2Writer.values();
  }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java

@@ -606,7 +606,7 @@ public Path commitDaughterRegion(final RegionInfo regionInfo, List<Path> allRegi
  writeRegionInfoFileContent(conf, fs, regionInfoFile, regionInfoContent);
  HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(
  env.getMasterConfiguration(), fs, getTableDir(), regionInfo, false);
-insertRegionFilesIntoStoreTracker(allRegionFiles, env, regionFs);
+ insertRegionFilesIntoStoreTracker(allRegionFiles, env, regionFs);
  }
  return regionDir;
  }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -432,6 +432,8 @@ public class HRegionServer extends HBaseServerBase<RSRpcServices>
  */
  final ServerNonceManager nonceManager;
 
+ private BrokenStoreFileCleaner brokenStoreFileCleaner;
+
  @InterfaceAudience.Private
  CompactedHFilesDischarger compactedFileDischarger;
 
@@ -1831,6 +1833,9 @@ private void startServices() throws IOException {
  if (this.slowLogTableOpsChore != null) {
  choreService.scheduleChore(slowLogTableOpsChore);
  }
+ if (this.brokenStoreFileCleaner != null) {
+ choreService.scheduleChore(brokenStoreFileCleaner);
+ }
 
  // Leases is not a Thread. Internally it runs a daemon thread. If it gets
  // an unhandled exception, it will just exit.
@@ -1910,6 +1915,22 @@ private void initializeThreads() {
  this.storefileRefresher = new StorefileRefresherChore(storefileRefreshPeriod,
  onlyMetaRefresh, this, this);
  }
+
+ int brokenStoreFileCleanerPeriod = conf.getInt(
+ BrokenStoreFileCleaner.BROKEN_STOREFILE_CLEANER_PERIOD,
+ BrokenStoreFileCleaner.DEFAULT_BROKEN_STOREFILE_CLEANER_PERIOD);
+ int brokenStoreFileCleanerDelay = conf.getInt(
+ BrokenStoreFileCleaner.BROKEN_STOREFILE_CLEANER_DELAY,
+ BrokenStoreFileCleaner.DEFAULT_BROKEN_STOREFILE_CLEANER_DELAY);
+ double brokenStoreFileCleanerDelayJitter = conf.getDouble(
+ BrokenStoreFileCleaner.BROKEN_STOREFILE_CLEANER_DELAY_JITTER,
+ BrokenStoreFileCleaner.DEFAULT_BROKEN_STOREFILE_CLEANER_DELAY_JITTER);
+ double jitterRate = (RandomUtils.nextDouble() - 0.5D) * brokenStoreFileCleanerDelayJitter;
+ long jitterValue = Math.round(brokenStoreFileCleanerDelay * jitterRate);
+ this.brokenStoreFileCleaner =
+ new BrokenStoreFileCleaner((int) (brokenStoreFileCleanerDelay + jitterValue),
+ brokenStoreFileCleanerPeriod, this, conf, this);
+
  registerConfigurationObservers();
  }
 
@@ -3484,6 +3505,11 @@ protected boolean clusterMode() {
  return !conf.getBoolean(MASTERLESS_CONFIG_NAME, false);
  }
 
+ @InterfaceAudience.Private
+ public BrokenStoreFileCleaner getBrokenStoreFileCleaner(){
+ return brokenStoreFileCleaner;
+ }
+
  @Override
  protected void stopChores() {
  shutdownChore(nonceManagerChore);
@@ -3494,5 +3520,6 @@ protected void stopChores() {
  shutdownChore(storefileRefresher);
  shutdownChore(fsUtilizationChore);
  shutdownChore(slowLogTableOpsChore);
+ shutdownChore(brokenStoreFileCleaner);
  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java

@@ -1157,6 +1157,12 @@ protected List<HStoreFile> doCompaction(CompactionRequestImpl cr,
  }
  }
  replaceStoreFiles(filesToCompact, sfs, true);
+
+ // This step is necessary for the correctness of BrokenStoreFileCleanerChore. It lets the
+ // CleanerChore know that compaction is done and the file can be cleaned up if compaction
+ // have failed.
+ storeEngine.resetCompactionWriter();
+
  if (cr.isMajor()) {
  majorCompactedCellsCount.addAndGet(getCompactionProgress().getTotalCompactingKVs());
  majorCompactedCellsSize.addAndGet(getCompactionProgress().totalCompactedSize);

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java

@@ -41,9 +41,11 @@
 import org.apache.hadoop.hbase.log.HBaseMarkers;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionPolicy;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
 import org.apache.hadoop.hbase.regionserver.compactions.Compactor;
 import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
 import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
+import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.ReflectionUtils;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -517,4 +519,23 @@ public void removeCompactedFiles(Collection<HStoreFile> compactedFiles) {
  throw new IOException("Unable to load configured store engine '" + className + "'", e);
  }
  }
+
+ /**
+ * Whether the implementation of the used storefile tracker requires you to write to temp
+ * directory first, i.e, does not allow broken store files under the actual data directory.
+ */
+ public boolean requireWritingToTmpDirFirst() {
+ return storeFileTracker.requireWritingToTmpDirFirst();
+ }
+
+ /**
+ * Resets the compaction writer when the new file is committed and used as active storefile.
+ * This step is necessary for the correctness of BrokenStoreFileCleanerChore. It lets the
+ * CleanerChore know that compaction is done and the file can be cleaned up if compaction
+ * have failed. Currently called in
+ * @see HStore#doCompaction(CompactionRequestImpl, Collection, User, long, List)
+ */
+ public void resetCompactionWriter(){
+ compactor.resetWriter();
+ }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeMultiFileWriter.java

@@ -58,7 +58,7 @@ public void setNoStripeMetadata() {
  }
 
  @Override
-protected Collection<StoreFileWriter> writers() {
+public Collection<StoreFileWriter> writers() {
  return existingWriters;
  }