Apache/hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-20401 Make MAX_WAIT and waitIfNotFinished in CleanerContext configurable

Browse Source 
Signed-off-by: Reid Chan <reidchan@apache.org>
This commit is contained in:
TAK LON WU 2018-07-22 21:16:45 -07:00 committed by Reid Chan
parent 896b69a0fc
commit 9db4a4059a
4 changed files with 138 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/HFileCleaner.java
View File

@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -77,6 +78,16 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
"hbase.regionserver.hfilecleaner.small.thread.count";
public final static int DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER = 1;
public static final String HFILE_DELETE_THREAD_TIMEOUT_MSEC =
"hbase.regionserver.hfilecleaner.thread.timeout.msec";
@VisibleForTesting
static final long DEFAULT_HFILE_DELETE_THREAD_TIMEOUT_MSEC = 60 * 1000L;
public static final String HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC =
"hbase.regionserver.hfilecleaner.thread.check.interval.msec";
@VisibleForTesting
static final long DEFAULT_HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC = 1000L;
private static final Log LOG = LogFactory.getLog(HFileCleaner.class);
StealJobQueue<HFileDeleteTask> largeFileQueue;
@ -86,6 +97,8 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
private int smallQueueInitSize;
private int largeFileDeleteThreadNumber;
private int smallFileDeleteThreadNumber;
private long cleanerThreadTimeoutMsec;
private long cleanerThreadCheckIntervalMsec;
private List<Thread> threads = new ArrayList<Thread>();
private boolean running;
@ -116,6 +129,11 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
conf.getInt(LARGE_HFILE_DELETE_THREAD_NUMBER, DEFAULT_LARGE_HFILE_DELETE_THREAD_NUMBER);
smallFileDeleteThreadNumber =
conf.getInt(SMALL_HFILE_DELETE_THREAD_NUMBER, DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER);
cleanerThreadTimeoutMsec =
conf.getLong(HFILE_DELETE_THREAD_TIMEOUT_MSEC, DEFAULT_HFILE_DELETE_THREAD_TIMEOUT_MSEC);
cleanerThreadCheckIntervalMsec =
conf.getLong(HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC,
DEFAULT_HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC);
startHFileDeleteThreads();
}
@ -147,7 +165,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
}
// wait for each submitted task to finish
for (HFileDeleteTask task : tasks) {
if (task.getResult()) {
if (task.getResult(cleanerThreadCheckIntervalMsec)) {
deletedFiles++;
}
}
@ -160,7 +178,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
* @return HFileDeleteTask to track progress
*/
private HFileDeleteTask deleteFile(FileStatus file) {
HFileDeleteTask task = new HFileDeleteTask(file);
HFileDeleteTask task = new HFileDeleteTask(file, cleanerThreadTimeoutMsec);
boolean enqueued = dispatch(task);
return enqueued ? task : null;
}
@ -299,17 +317,17 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
}
static class HFileDeleteTask implements Comparable<HFileDeleteTask> {
private static final long MAX_WAIT = 60 * 1000L;
private static final long WAIT_UNIT = 1000L;
boolean done = false;
boolean result;
final Path filePath;
final long fileLength;
final long timeoutMsec;
public HFileDeleteTask(FileStatus file) {
public HFileDeleteTask(FileStatus file, long timeoutMsec) {
this.filePath = file.getPath();
this.fileLength = file.getLen();
this.timeoutMsec = timeoutMsec;
}
public synchronized void setResult(boolean result) {
@ -318,17 +336,19 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
notify();
}
public synchronized boolean getResult() {
long waitTime = 0;
public synchronized boolean getResult(long waitIfNotFinished) {
long waitTimeMsec = 0;
try {
while (!done) {
wait(WAIT_UNIT);
waitTime += WAIT_UNIT;
long startTimeNanos = System.nanoTime();
wait(waitIfNotFinished);
waitTimeMsec += TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTimeNanos,
TimeUnit.NANOSECONDS);
if (done) {
return this.result;
}
if (waitTime > MAX_WAIT) {
LOG.warn("Wait more than " + MAX_WAIT + " ms for deleting " + this.filePath
if (waitTimeMsec > timeoutMsec) {
LOG.warn("Wait more than " + timeoutMsec + " ms for deleting " + this.filePath
+ ", exit...");
return false;
}
@ -397,6 +417,16 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
return throttlePoint;
}
@VisibleForTesting
long getCleanerThreadTimeoutMsec() {
return cleanerThreadTimeoutMsec;
}
@VisibleForTesting
long getCleanerThreadCheckIntervalMsec() {
return cleanerThreadCheckIntervalMsec;
}
@Override
public void onConfigurationChange(Configuration conf) {
if (!checkAndUpdateConfigurations(conf)) {
@ -469,6 +499,19 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> impleme
this.smallFileDeleteThreadNumber = smallFileDeleteThreadNumber;
updated = true;
}
long cleanerThreadTimeoutMsec =
conf.getLong(HFILE_DELETE_THREAD_TIMEOUT_MSEC, DEFAULT_HFILE_DELETE_THREAD_TIMEOUT_MSEC);
if (cleanerThreadTimeoutMsec != this.cleanerThreadTimeoutMsec) {
this.cleanerThreadTimeoutMsec = cleanerThreadTimeoutMsec;
updated = true;
}
long cleanerThreadCheckIntervalMsec =
conf.getLong(HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC,
DEFAULT_HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC);
if (cleanerThreadCheckIntervalMsec != this.cleanerThreadCheckIntervalMsec) {
this.cleanerThreadCheckIntervalMsec = cleanerThreadCheckIntervalMsec;
updated = true;
}
return updated;
}
}

65
hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/LogCleaner.java
View File

@ -26,6 +26,7 @@ import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -46,11 +47,24 @@ import org.apache.hadoop.hbase.wal.DefaultWALProvider;
public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
private static final Log LOG = LogFactory.getLog(LogCleaner.class.getName());
public static final String OLD_WALS_CLEANER_SIZE = "hbase.oldwals.cleaner.thread.size";
public static final int OLD_WALS_CLEANER_DEFAULT_SIZE = 2;
public static final String OLD_WALS_CLEANER_THREAD_SIZE = "hbase.oldwals.cleaner.thread.size";
public static final int DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE = 2;
public static final String OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC =
"hbase.oldwals.cleaner.thread.timeout.msec";
@VisibleForTesting
static final long DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC = 60 * 1000L;
public static final String OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC =
"hbase.oldwals.cleaner.thread.check.interval.msec";
@VisibleForTesting
static final long DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC = 500L;
private final LinkedBlockingQueue<CleanerContext> pendingDelete;
private List<Thread> oldWALsCleaner;
private long cleanerThreadTimeoutMsec;
private long cleanerThreadCheckIntervalMsec;
/**
* @param p the period of time to sleep between each run
@ -63,8 +77,12 @@ public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
Path oldLogDir) {
super("LogsCleaner", p, s, conf, fs, oldLogDir, HBASE_MASTER_LOGCLEANER_PLUGINS);
this.pendingDelete = new LinkedBlockingQueue<>();
int size = conf.getInt(OLD_WALS_CLEANER_SIZE, OLD_WALS_CLEANER_DEFAULT_SIZE);
int size = conf.getInt(OLD_WALS_CLEANER_THREAD_SIZE, DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE);
this.oldWALsCleaner = createOldWalsCleaner(size);
this.cleanerThreadTimeoutMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC);
this.cleanerThreadCheckIntervalMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC);
}
@Override
@ -76,7 +94,7 @@ public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
public void onConfigurationChange(Configuration conf) {
super.onConfigurationChange(conf);
int newSize = conf.getInt(OLD_WALS_CLEANER_SIZE, OLD_WALS_CLEANER_DEFAULT_SIZE);
int newSize = conf.getInt(OLD_WALS_CLEANER_THREAD_SIZE, DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE);
if (newSize == oldWALsCleaner.size()) {
if (LOG.isDebugEnabled()) {
LOG.debug("Size from configuration is the same as previous which is " +
@ -86,13 +104,18 @@ public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
}
interruptOldWALsCleaner();
oldWALsCleaner = createOldWalsCleaner(newSize);
cleanerThreadTimeoutMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC);
cleanerThreadCheckIntervalMsec = conf.getLong(OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC);
}
@Override
protected int deleteFiles(Iterable<FileStatus> filesToDelete) {
List<CleanerContext> results = new LinkedList<>();
for (FileStatus toDelete : filesToDelete) {
CleanerContext context = CleanerContext.createCleanerContext(toDelete);
CleanerContext context = CleanerContext.createCleanerContext(toDelete,
cleanerThreadTimeoutMsec);
if (context != null) {
pendingDelete.add(context);
results.add(context);
@ -101,7 +124,7 @@ public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
int deletedFiles = 0;
for (CleanerContext res : results) {
deletedFiles += res.getResult(500) ? 1 : 0;
deletedFiles += res.getResult(cleanerThreadCheckIntervalMsec) ? 1 : 0;
}
return deletedFiles;
}
@ -117,6 +140,16 @@ public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
return oldWALsCleaner.size();
}
@VisibleForTesting
long getCleanerThreadTimeoutMsec() {
return cleanerThreadTimeoutMsec;
}
@VisibleForTesting
long getCleanerThreadCheckIntervalMsec() {
return cleanerThreadCheckIntervalMsec;
}
private List<Thread> createOldWalsCleaner(int size) {
LOG.info("Creating OldWALs cleaners with size=" + size);
@ -190,20 +223,20 @@ public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
}
private static final class CleanerContext {
// At most waits 60 seconds
static final long MAX_WAIT = 60 * 1000;
final FileStatus target;
volatile boolean result;
volatile boolean setFromCleaner = false;
long timeoutMsec;
static CleanerContext createCleanerContext(FileStatus status) {
return status != null ? new CleanerContext(status) : null;
static CleanerContext createCleanerContext(FileStatus status, long timeoutMsec) {
return status != null ? new CleanerContext(status, timeoutMsec) : null;
}
private CleanerContext(FileStatus status) {
private CleanerContext(FileStatus status, long timeoutMsec) {
this.target = status;
this.result = false;
this.timeoutMsec = timeoutMsec;
}
synchronized void setResult(boolean res) {
@ -213,13 +246,15 @@ public class LogCleaner extends CleanerChore<BaseLogCleanerDelegate> {
}
synchronized boolean getResult(long waitIfNotFinished) {
long totalTime = 0;
long totalTimeMsec = 0;
try {
while (!setFromCleaner) {
long startTimeNanos = System.nanoTime();
wait(waitIfNotFinished);
totalTime += waitIfNotFinished;
if (totalTime >= MAX_WAIT) {
LOG.warn("Spend too much time to delete oldwals " + target);
totalTimeMsec += TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTimeNanos,
TimeUnit.NANOSECONDS);
if (totalTimeMsec >= timeoutMsec) {
LOG.warn("Spend too much time " + totalTimeMsec + " ms to delete oldwals " + target);
return result;
}
}

15
hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestHFileCleaner.java
View File

@ -36,10 +36,10 @@ import org.apache.hadoop.hbase.ChoreService;
import org.apache.hadoop.hbase.CoordinatedStateManager;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.ClusterConnection;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.EnvironmentEdge;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
@ -324,6 +324,8 @@ public class TestHFileCleaner {
final int SMALL_FILE_NUM = 20;
final int LARGE_THREAD_NUM = 2;
final int SMALL_THREAD_NUM = 4;
final long THREAD_TIMEOUT_MSEC = 30 * 1000L;
final long THREAD_CHECK_INTERVAL_MSEC = 500L;
Configuration conf = UTIL.getConfiguration();
// no cleaner policies = delete all files
@ -341,6 +343,10 @@ public class TestHFileCleaner {
Assert.assertEquals(ORIGINAL_THROTTLE_POINT, cleaner.getThrottlePoint());
Assert.assertEquals(ORIGINAL_QUEUE_INIT_SIZE, cleaner.getLargeQueueInitSize());
Assert.assertEquals(ORIGINAL_QUEUE_INIT_SIZE, cleaner.getSmallQueueInitSize());
Assert.assertEquals(HFileCleaner.DEFAULT_HFILE_DELETE_THREAD_TIMEOUT_MSEC,
cleaner.getCleanerThreadTimeoutMsec());
Assert.assertEquals(HFileCleaner.DEFAULT_HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC,
cleaner.getCleanerThreadCheckIntervalMsec());
// clean up archive directory and create files for testing
fs.delete(archivedHfileDir, true);
@ -368,6 +374,11 @@ public class TestHFileCleaner {
newConf.setInt(HFileCleaner.SMALL_HFILE_QUEUE_INIT_SIZE, UPDATE_QUEUE_INIT_SIZE);
newConf.setInt(HFileCleaner.LARGE_HFILE_DELETE_THREAD_NUMBER, LARGE_THREAD_NUM);
newConf.setInt(HFileCleaner.SMALL_HFILE_DELETE_THREAD_NUMBER, SMALL_THREAD_NUM);
newConf.setLong(HFileCleaner.HFILE_DELETE_THREAD_TIMEOUT_MSEC, THREAD_TIMEOUT_MSEC);
newConf.setLong(HFileCleaner.HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC,
THREAD_CHECK_INTERVAL_MSEC);
cleaner.onConfigurationChange(newConf);
LOG.debug("File deleted from large queue: " + cleaner.getNumOfDeletedLargeFiles()
+ "; from small queue: " + cleaner.getNumOfDeletedSmallFiles());
cleaner.onConfigurationChange(newConf);
@ -377,6 +388,8 @@ public class TestHFileCleaner {
Assert.assertEquals(UPDATE_QUEUE_INIT_SIZE, cleaner.getLargeQueueInitSize());
Assert.assertEquals(UPDATE_QUEUE_INIT_SIZE, cleaner.getSmallQueueInitSize());
Assert.assertEquals(LARGE_THREAD_NUM + SMALL_THREAD_NUM, cleaner.getCleanerThreads().size());
Assert.assertEquals(THREAD_TIMEOUT_MSEC, cleaner.getCleanerThreadTimeoutMsec());
Assert.assertEquals(THREAD_CHECK_INTERVAL_MSEC, cleaner.getCleanerThreadCheckIntervalMsec());
// make sure no cost when onConfigurationChange called with no change
List<Thread> oldThreads = cleaner.getCleanerThreads();

24
hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestLogsCleaner.java
View File

@ -242,7 +242,7 @@ public class TestLogsCleaner {
new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path("log1")),
new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path("log2"))
);
ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "testZooKeeperAbort-normal", null);
try {
cleaner.setConf(conf, zkw);
@ -261,14 +261,23 @@ public class TestLogsCleaner {
@Test
public void testOnConfigurationChange() throws Exception {
Configuration conf = TEST_UTIL.getConfiguration();
conf.setInt(LogCleaner.OLD_WALS_CLEANER_SIZE, LogCleaner.OLD_WALS_CLEANER_DEFAULT_SIZE);
conf.setInt(LogCleaner.OLD_WALS_CLEANER_THREAD_SIZE,
LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE);
conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC);
conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC);
// Prepare environments
Server server = new DummyServer();
Path oldWALsDir = new Path(TEST_UTIL.getDefaultRootDirPath(),
HConstants.HREGION_OLDLOGDIR_NAME);
FileSystem fs = TEST_UTIL.getDFSCluster().getFileSystem();
final LogCleaner cleaner = new LogCleaner(3000, server, conf, fs, oldWALsDir);
assertEquals(LogCleaner.OLD_WALS_CLEANER_DEFAULT_SIZE, cleaner.getSizeOfCleaners());
assertEquals(LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_SIZE, cleaner.getSizeOfCleaners());
assertEquals(LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC,
cleaner.getCleanerThreadTimeoutMsec());
assertEquals(LogCleaner.DEFAULT_OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
cleaner.getCleanerThreadCheckIntervalMsec());
// Create dir and files for test
fs.delete(oldWALsDir, true);
fs.mkdirs(oldWALsDir);
@ -287,9 +296,16 @@ public class TestLogsCleaner {
thread.start();
// change size of cleaners dynamically
int sizeToChange = 4;
conf.setInt(LogCleaner.OLD_WALS_CLEANER_SIZE, sizeToChange);
long threadTimeoutToChange = 30 * 1000L;
long threadCheckIntervalToChange = 250L;
conf.setInt(LogCleaner.OLD_WALS_CLEANER_THREAD_SIZE, sizeToChange);
conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_TIMEOUT_MSEC, threadTimeoutToChange);
conf.setLong(LogCleaner.OLD_WALS_CLEANER_THREAD_CHECK_INTERVAL_MSEC,
threadCheckIntervalToChange);
cleaner.onConfigurationChange(conf);
assertEquals(sizeToChange, cleaner.getSizeOfCleaners());
assertEquals(threadTimeoutToChange, cleaner.getCleanerThreadTimeoutMsec());
assertEquals(threadCheckIntervalToChange, cleaner.getCleanerThreadCheckIntervalMsec());
// Stop chore
thread.join();
status = fs.listStatus(oldWALsDir);