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

HBASE-19677 Miscellaneous HFileCleaner Improvements (BELUGA BEHR)

This commit is contained in:
tedyu 2017-12-31 09:47:12 -08:00
parent 0cd6050d09
commit 360d465a4a
1 changed files with 22 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -168,17 +168,13 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
if (task.fileLength >= this.throttlePoint) {
if (!this.largeFileQueue.offer(task)) {
// should never arrive here as long as we use PriorityQueue
if (LOG.isTraceEnabled()) {
LOG.trace("Large file deletion queue is full");
}
LOG.trace("Large file deletion queue is full");
return false;
}
} else {
if (!this.smallFileQueue.offer(task)) {
// should never arrive here as long as we use PriorityQueue
if (LOG.isTraceEnabled()) {
LOG.trace("Small file deletion queue is full");
}
LOG.trace("Small file deletion queue is full");
return false;
}
}
@ -208,7 +204,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
large.setDaemon(true);
large.setName(n + "-HFileCleaner.large." + i + "-" + System.currentTimeMillis());
large.start();
LOG.debug("Starting hfile cleaner for large files: " + large.getName());
LOG.debug("Starting hfile cleaner for large files: {}", large);
threads.add(large);
}
@ -223,7 +219,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
small.setDaemon(true);
small.setName(n + "-HFileCleaner.small." + i + "-" + System.currentTimeMillis());
small.start();
LOG.debug("Starting hfile cleaner for small files: " + small.getName());
LOG.debug("Starting hfile cleaner for small files: {}", small);
threads.add(small);
}
}
@ -235,20 +231,16 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
try {
task = queue.take();
} catch (InterruptedException e) {
if (LOG.isDebugEnabled()) {
LOG.debug("Interrupted while trying to take a task from queue", e);
}
LOG.debug("Interrupted while trying to take a task from queue", e);
break;
}
if (task != null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Removing: " + task.filePath + " from archive");
}
LOG.debug("Removing: {} from archive", task.filePath);
boolean succeed;
try {
succeed = this.fs.delete(task.filePath, false);
} catch (IOException e) {
LOG.warn("Failed to delete file " + task.filePath, e);
LOG.warn("Failed to delete file {}", task.filePath, e);
succeed = false;
}
task.setResult(succeed);
@ -258,9 +250,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
}
}
} finally {
if (LOG.isDebugEnabled()) {
LOG.debug("Exit thread: " + Thread.currentThread());
}
LOG.debug("Exit thread: {}", Thread.currentThread());
}
}
@ -275,9 +265,9 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
} else {
if (deletedSmallFiles.get() == Long.MAX_VALUE) {
LOG.info("Deleted more than Long.MAX_VALUE small files, reset counter to 0");
deletedSmallFiles.set(0L);;
deletedSmallFiles.set(0L);
}
if (fromLargeQueue && LOG.isTraceEnabled()) {
if (fromLargeQueue) {
LOG.trace("Stolen a small file deletion task in large file thread");
}
deletedSmallFiles.incrementAndGet();
@ -289,9 +279,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
*/
private void stopHFileDeleteThreads() {
running = false;
if (LOG.isDebugEnabled()) {
LOG.debug("Stopping file delete threads");
}
LOG.debug("Stopping file delete threads");
for(Thread thread: threads){
thread.interrupt();
}
@ -397,12 +385,8 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
// record the left over tasks
List<HFileDeleteTask> leftOverTasks =
new ArrayList<>(largeFileQueue.size() + smallFileQueue.size());
for (HFileDeleteTask task : largeFileQueue) {
leftOverTasks.add(task);
}
for (HFileDeleteTask task : smallFileQueue) {
leftOverTasks.add(task);
}
leftOverTasks.addAll(largeFileQueue);
leftOverTasks.addAll(smallFileQueue);
largeFileQueue = new StealJobQueue<>(largeQueueInitSize, smallQueueInitSize, COMPARATOR);
smallFileQueue = largeFileQueue.getStealFromQueue();
threads.clear();
@ -423,39 +407,39 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
int throttlePoint =
conf.getInt(HFILE_DELETE_THROTTLE_THRESHOLD, DEFAULT_HFILE_DELETE_THROTTLE_THRESHOLD);
if (throttlePoint != this.throttlePoint) {
LOG.debug("Updating throttle point, from " + this.throttlePoint + " to " + throttlePoint);
LOG.debug("Updating throttle point, from {} to {}", this.throttlePoint, throttlePoint);
this.throttlePoint = throttlePoint;
updated = true;
}
int largeQueueInitSize =
conf.getInt(LARGE_HFILE_QUEUE_INIT_SIZE, DEFAULT_LARGE_HFILE_QUEUE_INIT_SIZE);
if (largeQueueInitSize != this.largeQueueInitSize) {
LOG.debug("Updating largeQueueInitSize, from " + this.largeQueueInitSize + " to "
+ largeQueueInitSize);
LOG.debug("Updating largeQueueInitSize, from {} to {}", this.largeQueueInitSize,
largeQueueInitSize);
this.largeQueueInitSize = largeQueueInitSize;
updated = true;
}
int smallQueueInitSize =
conf.getInt(SMALL_HFILE_QUEUE_INIT_SIZE, DEFAULT_SMALL_HFILE_QUEUE_INIT_SIZE);
if (smallQueueInitSize != this.smallQueueInitSize) {
LOG.debug("Updating smallQueueInitSize, from " + this.smallQueueInitSize + " to "
+ smallQueueInitSize);
LOG.debug("Updating smallQueueInitSize, from {} to {}", this.smallQueueInitSize,
smallQueueInitSize);
this.smallQueueInitSize = smallQueueInitSize;
updated = true;
}
int largeFileDeleteThreadNumber =
conf.getInt(LARGE_HFILE_DELETE_THREAD_NUMBER, DEFAULT_LARGE_HFILE_DELETE_THREAD_NUMBER);
if (largeFileDeleteThreadNumber != this.largeFileDeleteThreadNumber) {
LOG.debug("Updating largeFileDeleteThreadNumber, from " + this.largeFileDeleteThreadNumber
+ " to " + largeFileDeleteThreadNumber);
LOG.debug("Updating largeFileDeleteThreadNumber, from {} to {}",
this.largeFileDeleteThreadNumber, largeFileDeleteThreadNumber);
this.largeFileDeleteThreadNumber = largeFileDeleteThreadNumber;
updated = true;
}
int smallFileDeleteThreadNumber =
conf.getInt(SMALL_HFILE_DELETE_THREAD_NUMBER, DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER);
if (smallFileDeleteThreadNumber != this.smallFileDeleteThreadNumber) {
LOG.debug("Updating smallFileDeleteThreadNumber, from " + this.smallFileDeleteThreadNumber
+ " to " + smallFileDeleteThreadNumber);
LOG.debug("Updating smallFileDeleteThreadNumber, from {} to {}",
this.smallFileDeleteThreadNumber, smallFileDeleteThreadNumber);
this.smallFileDeleteThreadNumber = smallFileDeleteThreadNumber;
updated = true;
}