You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2017/12/21 13:24:56 UTC
[07/42] hbase git commit: HBASE-19542 fix TestSafemodeBringsDownMaster
HBASE-19542 fix TestSafemodeBringsDownMaster
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/c811d7f9
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/c811d7f9
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/c811d7f9
Branch: refs/heads/HBASE-19397
Commit: c811d7f9655a1d424c8d4aeeaf4ef6fdb9d782b6
Parents: dc5ec06
Author: Chia-Ping Tsai <ch...@gmail.com>
Authored: Thu Dec 21 10:21:15 2017 +0800
Committer: Chia-Ping Tsai <ch...@gmail.com>
Committed: Thu Dec 21 10:32:14 2017 +0800
----------------------------------------------------------------------
.../hbase/regionserver/wal/AsyncFSWAL.java | 57 +++++++++++++-------
1 file changed, 39 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/c811d7f9/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java
index f8355e0..a9c440d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AsyncFSWAL.java
@@ -148,6 +148,10 @@ public class AsyncFSWAL extends AbstractFSWAL<AsyncWriter> {
"hbase.wal.async.use-shared-event-loop";
public static final boolean DEFAULT_ASYNC_WAL_USE_SHARED_EVENT_LOOP = false;
+ public static final String ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS =
+ "hbase.wal.async.wait.on.shutdown.seconds";
+ public static final int DEFAULT_ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS = 5;
+
private final EventLoopGroup eventLoopGroup;
private final ExecutorService consumeExecutor;
@@ -207,6 +211,8 @@ public class AsyncFSWAL extends AbstractFSWAL<AsyncWriter> {
private long highestProcessedAppendTxidAtLastSync;
+ private final int waitOnShutdownInSeconds;
+
public AsyncFSWAL(FileSystem fs, Path rootDir, String logDir, String archiveDir,
Configuration conf, List<WALActionsListener> listeners, boolean failIfWALExists,
String prefix, String suffix, EventLoopGroup eventLoopGroup,
@@ -254,6 +260,8 @@ public class AsyncFSWAL extends AbstractFSWAL<AsyncWriter> {
batchSize = conf.getLong(WAL_BATCH_SIZE, DEFAULT_WAL_BATCH_SIZE);
createMaxRetries =
conf.getInt(ASYNC_WAL_CREATE_MAX_RETRIES, DEFAULT_ASYNC_WAL_CREATE_MAX_RETRIES);
+ waitOnShutdownInSeconds = conf.getInt(ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS,
+ DEFAULT_ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS);
rollWriter();
}
@@ -699,30 +707,26 @@ public class AsyncFSWAL extends AbstractFSWAL<AsyncWriter> {
} finally {
consumeLock.unlock();
}
- long oldFileLen;
- if (oldWriter != null) {
- oldFileLen = oldWriter.getLength();
- closeExecutor.execute(() -> {
- try {
- oldWriter.close();
- } catch (IOException e) {
- LOG.warn("close old writer failed", e);
- }
- });
- } else {
- oldFileLen = 0L;
- }
- return oldFileLen;
+ return executeClose(closeExecutor, oldWriter);
}
@Override
protected void doShutdown() throws IOException {
waitForSafePoint();
- if (this.writer != null) {
- this.writer.close();
- this.writer = null;
- }
+ executeClose(closeExecutor, writer);
closeExecutor.shutdown();
+ try {
+ if (!closeExecutor.awaitTermination(waitOnShutdownInSeconds, TimeUnit.SECONDS)) {
+ LOG.error("We have waited " + waitOnShutdownInSeconds + " seconds but"
+ + " the close of async writer doesn't complete."
+ + "Please check the status of underlying filesystem"
+ + " or increase the wait time by the config \""
+ + ASYNC_WAL_WAIT_ON_SHUTDOWN_IN_SECONDS + "\"");
+ }
+ } catch (InterruptedException e) {
+ LOG.error("The wait for close of async writer is interrupted");
+ Thread.currentThread().interrupt();
+ }
IOException error = new IOException("WAL has been closed");
syncFutures.forEach(f -> f.done(f.getTxid(), error));
if (!(consumeExecutor instanceof EventLoop)) {
@@ -730,6 +734,23 @@ public class AsyncFSWAL extends AbstractFSWAL<AsyncWriter> {
}
}
+ private static long executeClose(ExecutorService closeExecutor, AsyncWriter writer) {
+ long fileLength;
+ if (writer != null) {
+ fileLength = writer.getLength();
+ closeExecutor.execute(() -> {
+ try {
+ writer.close();
+ } catch (IOException e) {
+ LOG.warn("close old writer failed", e);
+ }
+ });
+ } else {
+ fileLength = 0L;
+ }
+ return fileLength;
+ }
+
@Override
protected void doAppend(AsyncWriter writer, FSWALEntry entry) {
writer.append(entry);