Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ protected synchronized void addRollbackStep(Procedure<TEnvironment> proc) {
subprocStack = new ArrayList<>();
}
proc.addStackIndex(subprocStack.size());
LOG.debug("Add procedure {} as the {}th rollback step", proc, subprocStack.size());
LOG.trace("Add procedure {} as the {}th rollback step", proc, subprocStack.size());
subprocStack.add(proc);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
Expand Down Expand Up @@ -111,8 +113,13 @@ protected int deleteFiles(Iterable<FileStatus> filesToDelete) {
results.add(new CleanerContext(file));
}
}
if (results.isEmpty()) {
return 0;
}

LOG.debug("Old WAL files pending deletion: {}", results);
LOG.debug("Old WALs for delete: {}",
results.stream().map(cc -> cc.target.getPath().getName()).
collect(Collectors.joining(", ")));
pendingDelete.addAll(results);

int deletedFiles = 0;
Expand Down Expand Up @@ -140,7 +147,7 @@ long getCleanerThreadTimeoutMsec() {
}

private List<Thread> createOldWalsCleaner(int size) {
LOG.info("Creating {} OldWALs cleaner threads", size);
LOG.info("Creating {} old WALs cleaner threads", size);

List<Thread> oldWALsCleaner = new ArrayList<>(size);
for (int i = 0; i < size; i++) {
Expand Down Expand Up @@ -168,12 +175,12 @@ private void deleteFile() {
Preconditions.checkNotNull(context);
FileStatus oldWalFile = context.getTargetToClean();
try {
LOG.debug("Attempting to delete old WAL file: {}", oldWalFile);
LOG.debug("Deleting {}", oldWalFile);
boolean succeed = this.fs.delete(oldWalFile.getPath(), false);
context.setResult(succeed);
} catch (IOException e) {
// fs.delete() fails.
LOG.warn("Failed to clean old WAL file", e);
LOG.warn("Failed to delete old WAL file", e);
context.setResult(false);
}
} catch (InterruptedException ite) {
Expand All @@ -184,7 +191,7 @@ private void deleteFile() {
Thread.currentThread().interrupt();
return;
}
LOG.debug("Exiting");
LOG.trace("Exiting");
}
}

Expand Down Expand Up @@ -217,7 +224,7 @@ boolean getResult(long waitIfNotFinished) {
boolean completed = this.remainingResults.await(waitIfNotFinished,
TimeUnit.MILLISECONDS);
if (!completed) {
LOG.warn("Spend too much time [{}ms] to delete old WAL file: {}",
LOG.warn("Spent too much time [{}ms] deleting old WAL file: {}",
waitIfNotFinished, target);
return false;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -348,8 +348,8 @@ public void completionCleanup(final Procedure proc) {

private static <T extends Comparable<T>> void addToRunQueue(FairQueue<T> fairq, Queue<T> queue,
Supplier<String> reason) {
if (LOG.isDebugEnabled()) {
LOG.debug("Add {} to run queue because: {}", queue, reason.get());
if (LOG.isTraceEnabled()) {
LOG.trace("Add {} to run queue because: {}", queue, reason.get());
}
if (!AvlIterableList.isLinked(queue) && !queue.isEmpty()) {
fairq.add(queue);
Expand All @@ -358,8 +358,8 @@ private static <T extends Comparable<T>> void addToRunQueue(FairQueue<T> fairq,

private static <T extends Comparable<T>> void removeFromRunQueue(FairQueue<T> fairq,
Queue<T> queue, Supplier<String> reason) {
if (LOG.isDebugEnabled()) {
LOG.debug("Remove {} from run queue because: {}", queue, reason.get());
if (LOG.isTraceEnabled()) {
LOG.trace("Remove {} from run queue because: {}", queue, reason.get());
}
if (AvlIterableList.isLinked(queue)) {
fairq.remove(queue);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ protected void afterRoll(WAL wal) {
Path newFile = new Path(globalWALArchiveDir,
file.getName() + MasterProcedureUtil.ARCHIVED_PROC_WAL_SUFFIX);
if (fs.rename(file, newFile)) {
LOG.info("Successfully moved {} to {}", file, newFile);
LOG.info("Moved {} to {}", file, newFile);
} else {
LOG.warn("Failed to move archived wal from {} to global place {}", file, newFile);
}
Expand Down Expand Up @@ -124,4 +124,4 @@ static RegionProcedureStoreWALRoller create(Configuration conf, Abortable aborta
conf.setFloat(AbstractFSWAL.WAL_ROLL_MULTIPLIER, 0.5f);
return new RegionProcedureStoreWALRoller(conf, abortable, fs, walRootDir, globalWALRootDir);
}
}
}