diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java index 1b362af..4d4261e 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java @@ -43,7 +43,9 @@ import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -323,6 +325,30 @@ public class HBaseFsck extends Configured { this.executor = exec; } + private class FileLockCallable implements Callable { + @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 * @@ -331,31 +357,32 @@ public class HBaseFsck extends Configured { */ private FSDataOutputStream checkAndMarkRunningHbck() throws IOException { long start = EnvironmentEdgeManager.currentTime(); + + FileLockCallable callable = new FileLockCallable(); + ExecutorService executor = Executors.newFixedThreadPool(1); + FutureTask futureTask = new FutureTask(callable); + executor.execute(futureTask); + long duration = 0; + while (!futureTask.isDone() && duration <= 30000) { + Threads.sleep(10); + duration = EnvironmentEdgeManager.currentTime() - start; + } + if (!futureTask.isDone()) { + // took too long to obtain lock + LOG.warn("Took " + duration + " milliseconds to obtain lock"); + futureTask.cancel(true); + return null; + } + executor.shutdownNow(); + FSDataOutputStream stream = null; 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; - } - } finally { - long duration = EnvironmentEdgeManager.currentTime() - start; - if (duration > 30000) { - LOG.warn("Took " + duration + " milliseconds to obtain lock"); - // took too long to obtain lock - return null; - } + stream = futureTask.get(); + } catch (ExecutionException ee) { + LOG.warn("Encountered exception when opening lock file", ee); + } catch (InterruptedException ie) { + LOG.warn("Encountered exception when opening lock file", ie); } + return stream; } private void unlockHbck() {