You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by mb...@apache.org on 2012/05/02 20:27:01 UTC
svn commit: r1333143 - in /hbase/branches/0.89-fb/src:
main/java/org/apache/hadoop/hbase/
main/java/org/apache/hadoop/hbase/regionserver/
main/java/org/apache/hadoop/hbase/regionserver/wal/
main/java/org/apache/hadoop/hbase/util/ test/java/org/apache/h...
Author: mbautin
Date: Wed May 2 18:27:00 2012
New Revision: 1333143
URL: http://svn.apache.org/viewvc?rev=1333143&view=rev
Log:
[HBASE-5668][JIRA] HRegionServer.checkFileSystem() should only abort() after fs is down for some time
Author: pkhemani
Summary:
Introduce a timeout before checkFS decides to abort on failure. By default this timeout is set same as zookeeper.session.timeout
Test Plan: will test on titanmigrate002
Reviewers: kannan, amirshim, mbautin
Reviewed By: kannan
CC: hbase-eng@, hkuang
Differential Revision: https://phabricator.fb.com/D439401
Removed:
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/DroppedSnapshotException.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/wal/FailedLogCloseException.java
Modified:
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/SplitRequest.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/wal/HLog.java
hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java
hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerFileSystemFailure.java
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Wed May 2 18:27:00 2012
@@ -54,6 +54,7 @@ import java.util.concurrent.atomic.Atomi
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.avro.ipc.Server;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -62,7 +63,6 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.DoNotRetryIOException;
-import org.apache.hadoop.hbase.DroppedSnapshotException;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
@@ -1265,8 +1265,6 @@ public class HRegion implements HeapSize
* @return true if cache was flushed
*
* @throws IOException general io exceptions
- * @throws DroppedSnapshotException Thrown when replay of hlog is required
- * because a Snapshot was not properly persisted.
*/
public boolean flushcache() throws IOException {
MonitoredTask status = TaskMonitor.get().createStatus("Flushing " + this);
@@ -1346,8 +1344,6 @@ public class HRegion implements HeapSize
* @return true if the region needs compacting
*
* @throws IOException general io exceptions
- * @throws DroppedSnapshotException Thrown when replay of hlog is required
- * because a Snapshot was not properly persisted.
*/
protected boolean internalFlushcache(MonitoredTask status)
throws IOException {
@@ -1425,14 +1421,25 @@ public class HRegion implements HeapSize
// Otherwise, the snapshot content while backed up in the hlog, it will not
// be part of the current running servers state.
boolean compactionRequested = false;
- try {
- // A. Flush memstore to all the HStores.
- // Keep running vector of all store files that includes both old and the
- // just-made new flush store file.
+ // A. Flush memstore to all the HStores.
+ // Keep running vector of all store files that includes both old and the
+ // just-made new flush store file.
+ try {
for (StoreFlusher flusher : storeFlushers) {
flusher.flushCache(status);
}
+ } catch (IOException ioe) {
+ if (wal != null) {
+ wal.abortCacheFlush(this.regionInfo.getRegionName());
+ }
+ status.abort("Flush failed: " + StringUtils.stringifyException(ioe));
+ // The caller can recover from this IOException. No harm done if
+ // memstore flush fails.
+ throw ioe;
+ }
+
+ try {
Callable<Void> atomicWork = internalPreFlushcacheCommit();
@@ -1476,14 +1483,12 @@ public class HRegion implements HeapSize
// We used to only catch IOEs but its possible that we'd get other
// exceptions -- e.g. HBASE-659 was about an NPE -- so now we catch
// all and sundry.
- if (wal != null) {
- wal.abortCacheFlush(this.regionInfo.getRegionName());
- }
- DroppedSnapshotException dse = new DroppedSnapshotException("region: " +
- Bytes.toStringBinary(getRegionName()));
- dse.initCause(t);
- status.abort("Flush failed: " + StringUtils.stringifyException(t));
- throw dse;
+
+ // There is no recovery possible from failures in this block. Abort.
+ // But now that we have factored out memstore flushing in a separate
+ // block, there shouldn't really be any exception here.
+ LOG.fatal("failed to commit flushed memstore. Aborting.", t);
+ Runtime.getRuntime().halt(1);
}
// If we get to here, the HStores have been written. If we get an
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java Wed May 2 18:27:00 2012
@@ -279,6 +279,18 @@ public class HRegionServer implements HR
// Cache configuration and block cache reference
private final CacheConfig cacheConfig;
+
+ // prevents excessive checking of filesystem
+ private int minCheckFSIntervalMillis;
+
+ // abort if checkFS continues to fail for this long
+ private int checkFSAbortTimeOutMillis;
+
+ // time when last checkFS was done
+ private AtomicLong lastCheckFSAt = new AtomicLong(0);
+
+ // time since checkFS has been failing
+ private volatile long checkFSFailingSince;
private String stopReason = "not stopping";
@@ -336,6 +348,18 @@ public class HRegionServer implements HR
reinitialize();
SchemaMetrics.configureGlobally(conf);
cacheConfig = new CacheConfig(conf);
+
+ minCheckFSIntervalMillis =
+ conf.getInt("hbase.regionserver.min.check.fs.interval", 30000);
+ checkFSAbortTimeOutMillis =
+ conf.getInt("hbase.regionserver.check.fs.abort.timeout",
+ conf.getInt(HConstants.ZOOKEEPER_SESSION_TIMEOUT,
+ HConstants.DEFAULT_ZOOKEEPER_SESSION_TIMEOUT));
+ if (minCheckFSIntervalMillis > checkFSAbortTimeOutMillis) {
+ minCheckFSIntervalMillis = checkFSAbortTimeOutMillis;
+ }
+ LOG.info("minCheckFSIntervalMillis=" + minCheckFSIntervalMillis);
+ LOG.info("checkFSAbortTimeOutMillis=" + checkFSAbortTimeOutMillis);
}
/**
@@ -507,8 +531,8 @@ public class HRegionServer implements HR
if (!haveRootRegion.get()) {
HServerAddress rootServer = zooKeeperWrapper.readRootRegionLocation();
if (rootServer != null) {
- // By setting the root region location, we bypass the wait imposed on
- // HTable for all regions being assigned.
+ // By setting the root region location, we bypass the wait imposed
+ // on HTable for all regions being assigned.
this.connection.setRootRegionLocation(
new HRegionLocation(HRegionInfo.ROOT_REGIONINFO, rootServer));
haveRootRegion.set(true);
@@ -529,6 +553,7 @@ public class HRegionServer implements HR
for (HRegion r: onlineRegions.values()) {
hsl.addRegionInfo(createRegionLoad(r));
}
+ // XXX add a field in serverInfo to report to fsOK to master?
this.serverInfo.setLoad(hsl);
this.requestCount.set(0);
addOutboundMsgs(outboundMessages);
@@ -574,13 +599,11 @@ public class HRegionServer implements HR
break;
default:
- if (fsOk) {
- try {
- toDo.put(new ToDoEntry(msgs[i]));
- } catch (InterruptedException e) {
- throw new RuntimeException("Putting into msgQueue was " +
- "interrupted.", e);
- }
+ try {
+ toDo.put(new ToDoEntry(msgs[i]));
+ } catch (InterruptedException e) {
+ throw new RuntimeException("Putting into msgQueue was " +
+ "interrupted.", e);
}
}
}
@@ -975,23 +998,44 @@ public class HRegionServer implements HR
return stop;
}
-
/**
* Checks to see if the file system is still accessible.
* If not, sets abortRequested and stopRequested
*
* @return false if file system is not available
*/
- public boolean checkFileSystem() {
- if (this.fsOk && this.fs != null) {
+ public void checkFileSystem() {
+ long curtime = EnvironmentEdgeManager.currentTimeMillis();
+ synchronized (lastCheckFSAt){
+ if ((curtime - this.lastCheckFSAt.get()) <= this.minCheckFSIntervalMillis) {
+ return;
+ }
+ this.lastCheckFSAt.set(curtime);
+ }
+ if (this.fs != null) {
try {
FSUtils.checkFileSystemAvailable(this.fs, false);
+ this.fsOk = true;
+ return;
} catch (IOException e) {
- abort("File System not available", e);
+ if (this.fsOk) {
+ this.checkFSFailingSince = curtime;
+ }
this.fsOk = false;
+ // call abort immediately if checkFSAbortTimeOutMillis is 0
+ long timeToAbort = checkFSFailingSince + checkFSAbortTimeOutMillis -
+ curtime;
+ if (timeToAbort <= 0) {
+ abort("File System not available", e);
+ return;
+ }
+ LOG.warn("File System not available, will abort after " +
+ timeToAbort + "ms", e);
+ return;
}
}
- return this.fsOk;
+ this.fsOk = true;
+ return;
}
/*
@@ -1286,10 +1330,10 @@ public class HRegionServer implements HR
* Verify that server is healthy
*/
private boolean isHealthy() {
- if (!fsOk) {
- // File system problem
- return false;
- }
+ // Declare yourself healthy even if filesystem is not OK.
+ // Logic in checkFileSystem() and elsewhere now hopes that filesystem
+ // failures are transient.
+
// Verify that all threads are alive
if (!(leases.isAlive() &&
cacheFlusher.isAlive() && hlogRoller.isAlive() &&
@@ -1678,9 +1722,7 @@ public class HRegionServer implements HR
} else {
LOG.error("unable to process message" +
(e != null ? (": " + e.msg.toString()) : ""), ex);
- if (!checkFileSystem()) {
- break;
- }
+ checkFileSystem();
}
}
}
@@ -2735,6 +2777,10 @@ public class HRegionServer implements HR
throw new IOException("Server not running" +
(this.abortRequested? ", aborting": ""));
}
+ // its ok to call checkFileSystem() because it is rate limited
+ if (!fsOk) {
+ checkFileSystem();
+ }
if (!fsOk) {
throw new IOException("File system not available");
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java Wed May 2 18:27:00 2012
@@ -22,9 +22,11 @@ package org.apache.hadoop.hbase.regionse
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.RemoteExceptionHandler;
-import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
+import org.apache.hadoop.hbase.monitoring.MonitoredTask;
+import org.apache.hadoop.hbase.monitoring.TaskMonitor;
import org.apache.hadoop.hbase.regionserver.wal.LogRollListener;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.util.StringUtils;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -56,14 +58,23 @@ class LogRoller extends Thread implement
@Override
public void run() {
+ MonitoredTask status = null;
+ int retried = -1;
while (!server.isStopRequested()) {
long now = System.currentTimeMillis();
boolean periodic = false;
+ long modifiedRollPeriod;
if (!rollLog.get()) {
- periodic = (now - this.lastrolltime) > this.rollperiod;
+ if (server.getLog().isLogRollPending() == true) {
+ modifiedRollPeriod = server.threadWakeFrequency;
+ } else {
+ modifiedRollPeriod = this.rollperiod;
+ }
+ periodic = (now - this.lastrolltime) > modifiedRollPeriod;
if (!periodic) {
synchronized (rollLog) {
try {
+ // default 10s
rollLog.wait(server.threadWakeFrequency);
} catch (InterruptedException e) {
// Fall through
@@ -73,39 +84,49 @@ class LogRoller extends Thread implement
}
// Time for periodic roll
if (LOG.isDebugEnabled()) {
- LOG.debug("Hlog roll period " + this.rollperiod + "ms elapsed");
+ if (modifiedRollPeriod == this.rollperiod) {
+ LOG.debug("Hlog roll period " + this.rollperiod + "ms elapsed");
+ }
}
}
rollLock.lock(); // FindBugs UL_UNRELEASED_LOCK_EXCEPTION_PATH
try {
- this.lastrolltime = now;
byte [][] regionsToFlush = server.getLog().rollWriter();
+ if (status != null) {
+ status.markComplete("Log rolling succeeded after " + retried +
+ "retires.");
+ retried = -1;
+ status = null;
+ }
if (regionsToFlush != null) {
for (byte [] r: regionsToFlush) scheduleFlush(r);
}
- } catch (FailedLogCloseException e) {
- LOG.fatal("Forcing server shutdown", e);
- server.checkFileSystem();
- server.abort("Failed log close in log roller", e);
- } catch (java.net.ConnectException e) {
- LOG.fatal("Forcing server shutdown", e);
- server.checkFileSystem();
- server.abort("Failed connect in log roller", e);
} catch (IOException ex) {
- LOG.fatal("Log rolling failed with ioe: ",
- RemoteExceptionHandler.checkIOException(ex));
+ retried++;
+ String msg = "log roll failed." +
+ " retried=" + retried + ", " + StringUtils.stringifyException(ex);
+ if (status == null) {
+ LOG.warn("Log rolling failed with ioe. Will retry." +
+ " Will update status with exceptionif retry fails " +
+ RemoteExceptionHandler.checkIOException(ex));
+ status = TaskMonitor.get().createStatus(msg);
+ } else {
+ status.setStatus(msg);
+ }
server.checkFileSystem();
- // Abort if we get here. We probably won't recover an IOE. HBASE-1132
- server.abort("IOE in log roller", ex);
} catch (Exception ex) {
- LOG.error("Log rolling failed", ex);
- server.checkFileSystem();
- server.abort("Log rolling failed", ex);
+ LOG.fatal("Log rolling failed, unexpected exception. Force Aborting",
+ ex);
+ server.forceAbort();
} finally {
+ this.lastrolltime = System.currentTimeMillis();
rollLog.set(false);
rollLock.unlock();
}
}
+ if (status != null) {
+ status.abort("LogRoller exiting while log was unstable and roll pending");
+ }
LOG.info("LogRoller exiting.");
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java Wed May 2 18:27:00 2012
@@ -22,7 +22,6 @@ package org.apache.hadoop.hbase.regionse
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.DroppedSnapshotException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.RemoteExceptionHandler;
import org.apache.hadoop.hbase.util.Bytes;
@@ -144,7 +143,7 @@ class MemStoreFlusher extends Thread imp
continue;
}
if (!flushRegion(fqe)) {
- break;
+ LOG.warn("Failed to flush " + fqe.region);
}
} catch (InterruptedException ex) {
continue;
@@ -154,9 +153,7 @@ class MemStoreFlusher extends Thread imp
LOG.error("Cache flush failed" +
(fqe != null ? (" for region " + Bytes.toString(fqe.region.getRegionName())) : ""),
ex);
- if (!server.checkFileSystem()) {
- break;
- }
+ server.checkFileSystem();
}
}
this.regionsInQueue.clear();
@@ -261,21 +258,13 @@ class MemStoreFlusher extends Thread imp
server.compactSplitThread.requestCompaction(region, getName());
}
server.getMetrics().addFlush(region.getRecentFlushInfo());
- } catch (DroppedSnapshotException ex) {
- // Cache flush can fail in a few places. If it fails in a critical
- // section, we get a DroppedSnapshotException and a replay of hlog
- // is required. Currently the only way to do this is a restart of
- // the server. Abort because hdfs is probably bad (HBASE-644 is a case
- // where hdfs was bad but passed the hdfs check).
- server.abort("Replay of HLog required. Forcing server shutdown", ex);
- return false;
} catch (IOException ex) {
- LOG.error("Cache flush failed" +
- (region != null ? (" for region " + Bytes.toString(region.getRegionName())) : ""),
+ LOG.warn("Cache flush failed" +
+ (region != null ? (" for region " +
+ Bytes.toString(region.getRegionName())) : ""),
RemoteExceptionHandler.checkIOException(ex));
- if (!server.checkFileSystem()) {
- return false;
- }
+ server.checkFileSystem();
+ return false;
} finally {
lock.unlock();
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/SplitRequest.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/SplitRequest.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/SplitRequest.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/SplitRequest.java Wed May 2 18:27:00 2012
@@ -115,6 +115,8 @@ class SplitRequest implements Runnable {
LOG.error("Split failed " + this, RemoteExceptionHandler
.checkIOException(ex));
server.checkFileSystem();
+ // XXX should restructure the code above such that we don't have to abort
+ server.abort("Split failed", ex);
}
}
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/wal/HLog.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/wal/HLog.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/wal/HLog.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/regionserver/wal/HLog.java Wed May 2 18:27:00 2012
@@ -19,7 +19,7 @@
*/
package org.apache.hadoop.hbase.regionserver.wal;
-import static org.apache.hadoop.hbase.util.FSUtils.recoverFileLease;
+import static org.apache.hadoop.hbase.util.FSUtils.*;
import java.io.DataInput;
import java.io.DataOutput;
@@ -192,6 +192,7 @@ public class HLog implements Syncable {
* Current log file.
*/
Writer writer;
+ boolean writerCloseSyncDone = false;
/*
* Map of all log files but the current one.
@@ -244,6 +245,8 @@ public class HLog implements Syncable {
private final List<LogEntryVisitor> logEntryVisitors =
new CopyOnWriteArrayList<LogEntryVisitor>();
+ private boolean logRollPending = false;
+
/**
* Pattern used to validate a HLog file name
*/
@@ -482,7 +485,7 @@ public class HLog implements Syncable {
* @throws org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException
* @throws IOException
*/
- public byte [][] rollWriter() throws FailedLogCloseException, IOException {
+ public byte [][] rollWriter() throws IOException {
// Return if nothing to flush.
if (this.writer != null && this.numEntries.get() <= 0) {
return null;
@@ -494,6 +497,7 @@ public class HLog implements Syncable {
if (closed) {
return regionsToFlush;
}
+ this.logRollPending = true;
// Clean up current writer.
Path oldFile = cleanupCurrentWriter(this.filenum);
this.filenum = System.currentTimeMillis();
@@ -509,6 +513,11 @@ public class HLog implements Syncable {
this.hdfs_out =
((SequenceFileLogWriter)this.writer).getDFSCOutputStream();
}
+ // If there was an IOException before this point then either the old log
+ // has not been closed or the new writer has not been created.
+ // LogRoller will keep retyring. Nothing can be appended to the logs
+ // when logRollPending is true
+ this.logRollPending = false;
LOG.info((oldFile != null?
"Roll " + FSUtils.getPath(oldFile) + ", entries=" +
@@ -529,10 +538,13 @@ public class HLog implements Syncable {
// If so, then no new writes have come in since all regions were
// flushed (and removed from the lastSeqWritten map). Means can
// remove all but currently open log file.
- for (Map.Entry<Long, Path> e : this.outputfiles.entrySet()) {
- archiveLogFile(e.getValue(), e.getKey());
+ TreeSet<Long> tempSet = new TreeSet<Long>(outputfiles.keySet());
+ for (Long seqNum : tempSet) {
+ archiveLogFile(outputfiles.get(seqNum), seqNum);
+ outputfiles.remove(seqNum);
}
- this.outputfiles.clear();
+ assert outputfiles.size() == 0 :
+ "Someone added new log files? How?";
} else {
regionsToFlush = cleanOldLogs();
}
@@ -545,6 +557,10 @@ public class HLog implements Syncable {
return regionsToFlush;
}
+ public boolean isLogRollPending() {
+ return this.logRollPending;
+ }
+
/**
* Get a reader for the WAL.
* @param fs
@@ -623,7 +639,8 @@ public class HLog implements Syncable {
" from region " + Bytes.toString(oldestRegion));
}
for (Long seq : sequenceNumbers) {
- archiveLogFile(this.outputfiles.remove(seq), seq);
+ archiveLogFile(this.outputfiles.get(seq), seq);
+ this.outputfiles.remove(seq);
}
}
@@ -699,19 +716,28 @@ public class HLog implements Syncable {
throws IOException {
Path oldFile = null;
if (this.writer != null) {
- // Close the current writer, get a new one.
+ try {
+ if (!writerCloseSyncDone) {
+ this.writer.sync();
+ }
+ } catch (IOException ioe) {
+ LOG.fatal("log sync failed when trying to close " + this.writer);
+ Runtime.getRuntime().halt(1);
+ }
+ // Close the current writer
+ writerCloseSyncDone = true;
try {
this.writer.close();
- this.writer = null;
- } catch (IOException e) {
- // Failed close of log file. Means we're losing edits. For now,
- // shut ourselves down to minimize loss. Alternative is to try and
- // keep going. See HBASE-930.
- FailedLogCloseException flce =
- new FailedLogCloseException("#" + currentfilenum);
- flce.initCause(e);
- throw e;
+ } catch (IOException ioe) {
+ Path fname = computeFilename();
+ if (!tryRecoverFileLease(fs, fname, conf)) {
+ IOException ioe2 =
+ new IOException("lease recovery pending for " + fname, ioe);
+ throw ioe2;
+ }
}
+ writerCloseSyncDone = false;
+ this.writer = null;
if (currentfilenum >= 0) {
oldFile = computeFilename();
this.outputfiles.put(Long.valueOf(this.logSeqNum.get()), oldFile);
@@ -783,9 +809,7 @@ public class HLog implements Syncable {
if (LOG.isDebugEnabled()) {
LOG.debug("closing hlog writer in " + this.dir.toString());
}
- if (this.writer != null) {
- this.writer.close();
- }
+ cleanupCurrentWriter(-1);
}
} finally {
cacheFlushLock.unlock();
@@ -834,6 +858,9 @@ public class HLog implements Syncable {
if (this.closed) {
throw new IOException("Cannot append; log is closed");
}
+ if (this.logRollPending) {
+ throw new IOException("Cannot write; log unstable, roll pending");
+ }
long seqNum = obtainSeqNum();
logKey.setLogSeqNum(seqNum);
// The 'lastSeqWritten' map holds the sequence number of the oldest
@@ -886,6 +913,9 @@ public class HLog implements Syncable {
if (this.closed) {
throw new IOException("Cannot append; log is closed");
}
+ if (this.logRollPending) {
+ throw new IOException("Cannot append; log unstable, roll pending");
+ }
long seqNum = obtainSeqNum();
// The 'lastSeqWritten' map holds the sequence number of the oldest
// write for each region (i.e. the first edit added to the particular
@@ -1046,7 +1076,18 @@ public class HLog implements Syncable {
try {
long now = System.currentTimeMillis();
long doneUpto = this.unflushedEntries.get();
- this.writer.sync();
+ if (!logRollPending) {
+ // if logRollPending is true then we know that the data in the old
+ // log file has been synced. We know that no more appends were done
+ // after logRollPending became true. So syncing here is not
+ // required.
+ //
+ // If log roll is pending then either the old log file has been
+ // synced but not closed. Or old log file has been successfully
+ // closed but a new one has not yet been created. In the latter
+ // case this.writer will be NULL
+ this.writer.sync();
+ }
this.syncTillHere = doneUpto;
syncTime.inc(System.currentTimeMillis() - now);
this.forceSync = false;
@@ -1069,9 +1110,8 @@ public class HLog implements Syncable {
" still proceeding ahead...");
}
} catch (IOException e) {
- LOG.fatal("Could not append. Requesting close of hlog", e);
- requestLogRoll();
- throw e;
+ LOG.fatal("Could not sync hlog. Aborting", e);
+ Runtime.getRuntime().halt(1);
}
}
@@ -1226,24 +1266,32 @@ public class HLog implements Syncable {
* @param logSeqId
* @throws IOException
*/
- public void completeCacheFlush(final byte [] regionName, final byte [] tableName,
- final long logSeqId, final boolean isMetaRegion)
- throws IOException {
+ public void completeCacheFlush(final byte [] regionName,
+ final byte [] tableName,
+ final long logSeqId, final boolean isMetaRegion) {
try {
synchronized (updateLock) {
if (this.closed) {
return;
}
- long now = System.currentTimeMillis();
- WALEdit edit = completeCacheFlushLogEdit();
- HLogKey key = makeKey(regionName, tableName, logSeqId,
- System.currentTimeMillis());
- this.writer.append(new Entry(key, edit));
- this.numEntries.incrementAndGet();
+ try {
+ if (this.logRollPending) {
+ LOG.warn("Skipping write of cache-flush-done entry" +
+ " to unstable hlog. Log roll is pending");
+ return;
+ }
+ long now = System.currentTimeMillis();
+ WALEdit edit = completeCacheFlushLogEdit();
+ HLogKey key = makeKey(regionName, tableName, logSeqId,
+ System.currentTimeMillis());
+ this.writer.append(new Entry(key, edit));
+ this.numEntries.incrementAndGet();
+ } catch (IOException ioe) {
+ LOG.warn("Failed to write cache-flush-done entry. Ignoring.", ioe);
+ }
}
// sync txn to file system
this.sync(isMetaRegion);
-
} finally {
// updateLock not needed for removing snapshot's entry
// Cleaning up of lastSeqWritten is in the finally clause because we
Modified: hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java (original)
+++ hbase/branches/0.89-fb/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java Wed May 2 18:27:00 2012
@@ -616,23 +616,24 @@ public class FSUtils {
}
- /*
- * Recover file lease. Used when a file might be suspect to be had been left open by another process. <code>p</code>
+ /**
+ *
* @param fs
* @param p
- * @param append True if append supported
+ * @param conf
+ * @return true if lease is recoverd, false if lease recovery is pending
* @throws IOException
*/
- public static void recoverFileLease(final FileSystem fs, final Path p, Configuration conf)
- throws IOException{
- if (!isAppendSupported(conf)) {
+ public static boolean tryRecoverFileLease(final FileSystem fs, final Path p,
+ Configuration conf) throws IOException {
+ if (!isAppendSupported(conf)) {
LOG.warn("Running on HDFS without append enabled may result in data loss");
- return;
+ return true;
}
// lease recovery not needed for local file system case.
// currently, local file system doesn't implement append either.
if (!(fs instanceof DistributedFileSystem)) {
- return;
+ return true;
}
long startWaiting = System.currentTimeMillis();
@@ -641,55 +642,64 @@ public class FSUtils {
LOG.info("Recovering file " + p + ", discard last block: "
+ discardlastBlock);
- // Trying recovery
- boolean recovered = false;
- while (!recovered) {
+ try {
try {
- try {
- if (fs instanceof DistributedFileSystem) {
- DistributedFileSystem dfs = (DistributedFileSystem)fs;
- try {
- DistributedFileSystem.class.getMethod("recoverLease",
- new Class[] {Path.class, Boolean.class}).
- invoke(dfs, p, new Boolean(discardlastBlock));
- } catch (NoSuchMethodException nsme) {
- DistributedFileSystem.class.getMethod("recoverLease",
- new Class[] {Path.class}).invoke(dfs, p);
- }
- } else {
- throw new Exception("Not a DistributedFileSystem");
- }
- } catch (InvocationTargetException ite) {
- // function was properly called, but threw it's own exception
- throw (IOException) ite.getCause();
- } catch (Exception e) {
- LOG.debug("Failed fs.recoverLease invocation, " + e.toString() +
- ", trying fs.append instead");
- FSDataOutputStream out = fs.append(p);
- out.close();
- }
- recovered = true;
- } catch (IOException e) {
- e = RemoteExceptionHandler.checkIOException(e);
- if (e instanceof AlreadyBeingCreatedException) {
- // We expect that we'll get this message while the lease is still
- // within its soft limit, but if we get it past that, it means
- // that the RS is holding onto the file even though it lost its
- // znode. We could potentially abort after some time here.
- long waitedFor = System.currentTimeMillis() - startWaiting;
- if (waitedFor > FSConstants.LEASE_SOFTLIMIT_PERIOD) {
- LOG.warn("Waited " + waitedFor + "ms for lease recovery on " + p +
- ":" + e.getMessage());
+ if (fs instanceof DistributedFileSystem) {
+ DistributedFileSystem dfs = (DistributedFileSystem)fs;
+ try {
+ return (Boolean) DistributedFileSystem.class.getMethod("recoverLease",
+ new Class[] {Path.class, Boolean.class}).
+ invoke(dfs, p, new Boolean(discardlastBlock));
+ } catch (NoSuchMethodException nsme) {
+ return (Boolean) DistributedFileSystem.class.getMethod("recoverLease",
+ new Class[] {Path.class}).invoke(dfs, p);
}
- } else if (e instanceof LeaseExpiredException &&
- e.getMessage().contains("File does not exist")) {
- // This exception comes out instead of FNFE, fix it
- throw new FileNotFoundException(
- "The given HLog wasn't found at " + p.toString());
} else {
- throw new IOException("Failed to open " + p + " for append", e);
+ throw new Exception("Not a DistributedFileSystem");
}
+ } catch (InvocationTargetException ite) {
+ // function was properly called, but threw it's own exception
+ throw (IOException) ite.getCause();
+ } catch (Exception e) {
+ LOG.debug("Failed fs.recoverLease invocation, " + e.toString() +
+ ", trying fs.append instead");
+ FSDataOutputStream out = fs.append(p);
+ out.close();
+ }
+ return true;
+ } catch (IOException e) {
+ e = RemoteExceptionHandler.checkIOException(e);
+ if (e instanceof AlreadyBeingCreatedException) {
+ // We expect that we'll get this message while the lease is still
+ // within its soft limit, but if we get it past that, it means
+ // that the RS is holding onto the file even though it lost its
+ // znode. We could potentially abort after some time here.
+ long waitedFor = System.currentTimeMillis() - startWaiting;
+ if (waitedFor > FSConstants.LEASE_SOFTLIMIT_PERIOD) {
+ LOG.warn("Waited " + waitedFor + "ms for lease recovery on " + p +
+ ":" + e.getMessage());
+ }
+ } else if (e instanceof LeaseExpiredException &&
+ e.getMessage().contains("File does not exist")) {
+ // This exception comes out instead of FNFE, fix it
+ throw new FileNotFoundException(
+ "The given HLog wasn't found at " + p.toString());
+ } else {
+ throw new IOException("Failed to open " + p + " for append", e);
}
+ }
+ return false;
+ }
+
+ /*
+ * Recover file lease. Used when a file might be suspect to be had been left open by another process. <code>p</code>
+ * @param fs
+ * @param p
+ * @throws IOException
+ */
+ public static void recoverFileLease(final FileSystem fs, final Path p,
+ Configuration conf) throws IOException {
+ while (!tryRecoverFileLease(fs, p, conf)) {
try {
Thread.sleep(1000);
} catch (InterruptedException ex) {
@@ -700,6 +710,7 @@ public class FSUtils {
LOG.info("Finished lease recover attempt for " + p);
}
+
/**
* Runs through the HBase rootdir and creates a reverse lookup map for
* table StoreFile names to the full Path.
@@ -1139,4 +1150,4 @@ class FSRegionScanner implements Runnabl
LOG.warn("Problem scanning file system", e);
}
}
-}
\ No newline at end of file
+}
Modified: hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerFileSystemFailure.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerFileSystemFailure.java?rev=1333143&r1=1333142&r2=1333143&view=diff
==============================================================================
--- hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerFileSystemFailure.java (original)
+++ hbase/branches/0.89-fb/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerFileSystemFailure.java Wed May 2 18:27:00 2012
@@ -88,7 +88,8 @@ public class TestHRegionServerFileSystem
for(RegionServerThread serverThread : servers) {
HRegionServer server = serverThread.getRegionServer();
if (serverThread.isAlive() && !server.isStopRequested()) {
- assertFalse(server.checkFileSystem());
+ server.checkFileSystem();
+ assertFalse(server.fsOk);
break;
}
}