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

HBASE-12785 Use FutureTask to timeout the attempt to get the lock for hbck

This commit is contained in:
tedyu 2015-01-05 14:10:48 -08:00
parent e7795007b0
commit f9873622b8
1 changed files with 46 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
View File

@ -45,8 +45,12 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
@ -323,6 +327,30 @@ public class HBaseFsck extends Configured implements Closeable {
this.executor = exec; this.executor = exec;
} }
private class FileLockCallable implements Callable<FSDataOutputStream> {
@Override
public FSDataOutputStream call() throws IOException {
try {
FileSystem fs = FSUtils.getCurrentFileSystem(getConf());
FsPermission defaultPerms = FSUtils.getFilePermissions(fs, getConf(),
HConstants.DATA_FILE_UMASK_KEY);
Path tmpDir = new Path(FSUtils.getRootDir(getConf()), HConstants.HBASE_TEMP_DIRECTORY);
fs.mkdirs(tmpDir);
HBCK_LOCK_PATH = new Path(tmpDir, HBCK_LOCK_FILE);
final FSDataOutputStream out = FSUtils.create(fs, HBCK_LOCK_PATH, defaultPerms, false);
out.writeBytes(InetAddress.getLocalHost().toString());
out.flush();
return out;
} catch(RemoteException e) {
if(AlreadyBeingCreatedException.class.getName().equals(e.getClassName())){
return null;
} else {
throw e;
}
}
}
}
/** /**
* This method maintains a lock using a file. If the creation fails we return null * This method maintains a lock using a file. If the creation fails we return null
* *
@ -330,32 +358,27 @@ public class HBaseFsck extends Configured implements Closeable {
* @throws IOException * @throws IOException
*/ */
private FSDataOutputStream checkAndMarkRunningHbck() throws IOException { private FSDataOutputStream checkAndMarkRunningHbck() throws IOException {
long start = EnvironmentEdgeManager.currentTime(); FileLockCallable callable = new FileLockCallable();
ExecutorService executor = Executors.newFixedThreadPool(1);
FutureTask<FSDataOutputStream> futureTask = new FutureTask<FSDataOutputStream>(callable);
executor.execute(futureTask);
final int timeoutInSeconds = 30;
FSDataOutputStream stream = null;
try { try {
FileSystem fs = FSUtils.getCurrentFileSystem(getConf()); stream = futureTask.get(30, TimeUnit.SECONDS);
FsPermission defaultPerms = FSUtils.getFilePermissions(fs, getConf(), } catch (ExecutionException ee) {
HConstants.DATA_FILE_UMASK_KEY); LOG.warn("Encountered exception when opening lock file", ee);
Path tmpDir = new Path(FSUtils.getRootDir(getConf()), HConstants.HBASE_TEMP_DIRECTORY); } catch (InterruptedException ie) {
fs.mkdirs(tmpDir); LOG.warn("Interrupted when opening lock file", ie);
HBCK_LOCK_PATH = new Path(tmpDir, HBCK_LOCK_FILE); Thread.currentThread().interrupt();
final FSDataOutputStream out = FSUtils.create(fs, HBCK_LOCK_PATH, defaultPerms, false); } catch (TimeoutException exception) {
out.writeBytes(InetAddress.getLocalHost().toString()); // took too long to obtain lock
out.flush(); LOG.warn("Took more than " + timeoutInSeconds + " seconds in obtaining lock");
return out; futureTask.cancel(true);
} catch(RemoteException e) {
if(AlreadyBeingCreatedException.class.getName().equals(e.getClassName())){
return null;
} else {
throw e;
}
} finally { } finally {
long duration = EnvironmentEdgeManager.currentTime() - start; executor.shutdownNow();
if (duration > 30000) {
LOG.warn("Took " + duration + " milliseconds to obtain lock");
// took too long to obtain lock
return null;
}
} }
return stream;
} }
private void unlockHbck() { private void unlockHbck() {