HBASE-20902 when WAL sync failed, we should bypass the failed DN that previously used

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/wal/MetricsWALSource.java

@@ -64,6 +64,9 @@ public interface MetricsWALSource extends BaseSource {
   String LOW_REPLICA_ROLL_REQUESTED = "lowReplicaRollRequest";
   String LOW_REPLICA_ROLL_REQUESTED_DESC =
       "How many times a log roll was requested due to too few DN's in the write pipeline.";
+  String SYNC_FAILED_ROLL_REQUESTED = "syncFailedRollRequest";
+  String SYNC_FAILED_ROLL_REQUESTED_DESC =
+      "How many times a log roll was requested due to sync failed.";
   String WRITTEN_BYTES = "writtenBytes";
   String WRITTEN_BYTES_DESC = "Size (in bytes) of the data written to the WAL.";
 
@@ -96,6 +99,8 @@ public interface MetricsWALSource extends BaseSource {
 
   void incrementLowReplicationLogRoll();
 
+  void incrementSyncFailedLogRoll();
+
   long getSlowAppendCount();
 
   void incrementWrittenBytes(long val);

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/regionserver/wal/MetricsWALSourceImpl.java

@@ -39,6 +39,7 @@ public class MetricsWALSourceImpl extends BaseSourceImpl implements MetricsWALSo
   private final MutableFastCounter slowAppendCount;
   private final MutableFastCounter logRollRequested;
   private final MutableFastCounter lowReplicationLogRollRequested;
+  private final MutableFastCounter syncFailedLogRollRequested;
   private final MutableFastCounter writtenBytes;
 
   public MetricsWALSourceImpl() {
@@ -62,6 +63,8 @@ public MetricsWALSourceImpl(String metricsName,
         this.getMetricsRegistry().newCounter(ROLL_REQUESTED, ROLL_REQUESTED_DESC, 0L);
     lowReplicationLogRollRequested = this.getMetricsRegistry()
         .newCounter(LOW_REPLICA_ROLL_REQUESTED, LOW_REPLICA_ROLL_REQUESTED_DESC, 0L);
+    syncFailedLogRollRequested = this.getMetricsRegistry()
+        .newCounter(SYNC_FAILED_ROLL_REQUESTED, SYNC_FAILED_ROLL_REQUESTED_DESC, 0L);
     writtenBytes = this.getMetricsRegistry().newCounter(WRITTEN_BYTES, WRITTEN_BYTES_DESC, 0L);
   }
 
@@ -100,6 +103,11 @@ public void incrementLowReplicationLogRoll() {
     lowReplicationLogRollRequested.incr();
   }
 
+  @Override
+  public void incrementSyncFailedLogRoll() {
+    syncFailedLogRollRequested.incr();
+  }
+
   @Override
   public long getSlowAppendCount() {
     return slowAppendCount.value();

hbase-server/src/main/java/org/apache/hadoop/hbase/io/asyncfs/AsyncFSOutputHelper.java

@@ -43,12 +43,12 @@ private AsyncFSOutputHelper() {
    * Create {@link FanOutOneBlockAsyncDFSOutput} for {@link DistributedFileSystem}, and a simple
    * implementation for other {@link FileSystem} which wraps around a {@link FSDataOutputStream}.
    */
-  public static AsyncFSOutput createOutput(FileSystem fs, Path f, boolean overwrite,
+  public static AsyncFSOutput createOutput(FileSystem fs, Path f, Path oldPath, boolean overwrite,
       boolean createParent, short replication, long blockSize, EventLoopGroup eventLoopGroup,
       Class<? extends Channel> channelClass)
       throws IOException, CommonFSUtils.StreamLacksCapabilityException {
     if (fs instanceof DistributedFileSystem) {
-      return FanOutOneBlockAsyncDFSOutputHelper.createOutput((DistributedFileSystem) fs, f,
+      return FanOutOneBlockAsyncDFSOutputHelper.createOutput((DistributedFileSystem) fs, f, oldPath,
         overwrite, createParent, replication, blockSize, eventLoopGroup, channelClass);
     }
     final FSDataOutputStream out;

hbase-server/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutput.java

@@ -18,11 +18,9 @@
 package org.apache.hadoop.hbase.io.asyncfs;
 
 import static org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputHelper.HEART_BEAT_SEQNO;
-import static org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputHelper.READ_TIMEOUT;
 import static org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputHelper.completeFile;
 import static org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputHelper.endFileLease;
 import static org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputHelper.getStatus;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY;
 import static org.apache.hbase.thirdparty.io.netty.handler.timeout.IdleState.READER_IDLE;
 import static org.apache.hbase.thirdparty.io.netty.handler.timeout.IdleState.WRITER_IDLE;
 
@@ -44,6 +42,7 @@
 import org.apache.hadoop.crypto.Encryptor;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputHelper.CancelOnClose;
+import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
 import org.apache.hadoop.hbase.util.CancelableProgressable;
 import org.apache.hadoop.hbase.util.FSUtils;
 import org.apache.hadoop.hdfs.DFSClient;
@@ -346,7 +345,7 @@ private void setupReceiver(int timeoutMs) {
     this.alloc = alloc;
     this.buf = alloc.directBuffer(sendBufSizePRedictor.initialSize());
     this.state = State.STREAMING;
-    setupReceiver(conf.getInt(DFS_CLIENT_SOCKET_TIMEOUT_KEY, READ_TIMEOUT));
+    setupReceiver(conf.getInt(AbstractFSWAL.WAL_SYNC_TIMEOUT_MS, AbstractFSWAL.DEFAULT_WAL_SYNC_TIMEOUT_MS));
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutputHelper.java

@@ -48,6 +48,7 @@
 import org.apache.hadoop.fs.UnresolvedLinkException;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hbase.client.ConnectionUtils;
+import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
 import org.apache.hadoop.hbase.util.CancelableProgressable;
 import org.apache.hadoop.hbase.util.FSUtils;
 import org.apache.hadoop.hdfs.DFSClient;
@@ -682,7 +683,7 @@ private static List<Future<Channel>> connectToDataNodes(Configuration conf, DFSC
     DatanodeInfo[] datanodeInfos = locatedBlock.getLocations();
     boolean connectToDnViaHostname =
         conf.getBoolean(DFS_CLIENT_USE_DN_HOSTNAME, DFS_CLIENT_USE_DN_HOSTNAME_DEFAULT);
-    int timeoutMs = conf.getInt(DFS_CLIENT_SOCKET_TIMEOUT_KEY, READ_TIMEOUT);
+    int timeoutMs = conf.getInt(AbstractFSWAL.WAL_SYNC_TIMEOUT_MS, AbstractFSWAL.DEFAULT_WAL_SYNC_TIMEOUT_MS);
     ExtendedBlock blockCopy = new ExtendedBlock(locatedBlock.getBlock());
     blockCopy.setNumBytes(locatedBlock.getBlockSize());
     ClientOperationHeaderProto header = ClientOperationHeaderProto.newBuilder()
@@ -740,7 +741,7 @@ public NameNodeException(Throwable cause) {
     }
   }
 
-  private static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem dfs, String src,
+  private static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem dfs, String src, Path oldPath,
       boolean overwrite, boolean createParent, short replication, long blockSize,
       EventLoopGroup eventLoopGroup, Class<? extends Channel> channelClass) throws IOException {
     Configuration conf = dfs.getConf();
@@ -751,6 +752,24 @@ private static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem d
     int createMaxRetries = conf.getInt(ASYNC_DFS_OUTPUT_CREATE_MAX_RETRIES,
       DEFAULT_ASYNC_DFS_OUTPUT_CREATE_MAX_RETRIES);
     DatanodeInfo[] excludesNodes = EMPTY_DN_ARRAY;
+    if (oldPath != null) {
+      String oldPathStr = oldPath.toUri().getPath();
+      long len = namenode.getFileInfo(oldPathStr).getLen();
+      for(LocatedBlock block : namenode.getBlockLocations(oldPathStr, Math.max(0, len - 1), len)
+          .getLocatedBlocks()) {
+        for(DatanodeInfo dn : block.getLocations()) {
+          excludesNodes = ArrayUtils.add(excludesNodes, dn);
+        }
+      }
+      if (LOG.isDebugEnabled()) {
+        StringBuilder sb = new StringBuilder("create new output because old wal sync failed, old path is: ");
+    	sb.append(oldPathStr).append(", newPath excludesNodes are :");
+    	for(DatanodeInfo info : excludesNodes) {
+    	  sb.append(info.getInfoAddr()).append(";");
+    	}
+    	LOG.debug(sb.toString());
+      }
+    }
     for (int retry = 0;; retry++) {
       HdfsFileStatus stat;
       try {
@@ -837,15 +856,15 @@ public void operationComplete(Future<Channel> future) throws Exception {
    * Create a {@link FanOutOneBlockAsyncDFSOutput}. The method maybe blocked so do not call it
    * inside an {@link EventLoop}.
    */
-  public static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem dfs, Path f,
+  public static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem dfs, Path f, Path oldPath,
       boolean overwrite, boolean createParent, short replication, long blockSize,
       EventLoopGroup eventLoopGroup, Class<? extends Channel> channelClass) throws IOException {
     return new FileSystemLinkResolver<FanOutOneBlockAsyncDFSOutput>() {
 
       @Override
       public FanOutOneBlockAsyncDFSOutput doCall(Path p)
           throws IOException, UnresolvedLinkException {
-        return createOutput(dfs, p.toUri().getPath(), overwrite, createParent, replication,
+        return createOutput(dfs, p.toUri().getPath(), oldPath, overwrite, createParent, replication,
           blockSize, eventLoopGroup, channelClass);
       }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java

@@ -35,6 +35,7 @@
 import org.apache.hadoop.hbase.regionserver.wal.WALClosedException;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.HasThread;
+import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hadoop.hbase.wal.WAL;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -57,7 +58,8 @@ public class LogRoller extends HasThread implements Closeable {
   private static final Logger LOG = LoggerFactory.getLogger(LogRoller.class);
   private final ReentrantLock rollLock = new ReentrantLock();
   private final AtomicBoolean rollLog = new AtomicBoolean(false);
-  private final ConcurrentHashMap<WAL, Boolean> walNeedsRoll = new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<WAL, Pair<Boolean,Boolean>> walNeedsRoll =
+      new ConcurrentHashMap<WAL, Pair<Boolean, Boolean>>();
   private final Server server;
   protected final RegionServerServices services;
   private volatile long lastrolltime = System.currentTimeMillis();
@@ -70,11 +72,15 @@ public class LogRoller extends HasThread implements Closeable {
   private volatile boolean running = true;
 
   public void addWAL(final WAL wal) {
-    if (null == walNeedsRoll.putIfAbsent(wal, Boolean.FALSE)) {
+    if (null == walNeedsRoll.putIfAbsent(wal, new Pair<Boolean,Boolean>(Boolean.FALSE, Boolean.FALSE))) {
       wal.registerWALActionsListener(new WALActionsListener() {
         @Override
-        public void logRollRequested(boolean lowReplicas) {
-          walNeedsRoll.put(wal, Boolean.TRUE);
+        public void logRollRequested(boolean lowReplicas, boolean syncFaild) {
+          Pair<Boolean,Boolean> walInfo = walNeedsRoll.get(wal);
+          walInfo.setFirst(Boolean.TRUE);
+          if (syncFaild) {
+            walInfo.setSecond(Boolean.TRUE);
+          }
           // TODO logs will contend with each other here, replace with e.g. DelayedQueue
           synchronized(rollLog) {
             rollLog.set(true);
@@ -87,7 +93,7 @@ public void logRollRequested(boolean lowReplicas) {
 
   public void requestRollAll() {
     for (WAL wal : walNeedsRoll.keySet()) {
-      walNeedsRoll.put(wal, Boolean.TRUE);
+      walNeedsRoll.put(wal, new Pair<Boolean, Boolean>(Boolean.TRUE, Boolean.FALSE));
     }
     synchronized(rollLog) {
       rollLog.set(true);
@@ -122,9 +128,9 @@ public void interrupt() {
    */
   void checkLowReplication(long now) {
     try {
-      for (Entry<WAL, Boolean> entry : walNeedsRoll.entrySet()) {
+      for (Entry<WAL, Pair<Boolean, Boolean>> entry : walNeedsRoll.entrySet()) {
         WAL wal = entry.getKey();
-        boolean needRollAlready = entry.getValue();
+        boolean needRollAlready = entry.getValue().getFirst();
         if (needRollAlready || !(wal instanceof AbstractFSWAL)) {
           continue;
         }
@@ -180,16 +186,18 @@ public void run() {
       rollLock.lock(); // FindBugs UL_UNRELEASED_LOCK_EXCEPTION_PATH
       try {
         this.lastrolltime = now;
-        for (Iterator<Entry<WAL, Boolean>> iter = walNeedsRoll.entrySet().iterator(); iter
+        for (Iterator<Entry<WAL, Pair<Boolean,Boolean>>> iter = walNeedsRoll.entrySet().iterator(); iter
             .hasNext();) {
-          Entry<WAL, Boolean> entry = iter.next();
+          Entry<WAL, Pair<Boolean,Boolean>> entry = iter.next();
           final WAL wal = entry.getKey();
+          Pair<Boolean, Boolean> walInfo = entry.getValue();
+          boolean syncFailed = walInfo.getSecond().booleanValue();
           // Force the roll if the logroll.period is elapsed or if a roll was requested.
           // The returned value is an array of actual region names.
           try {
-            final byte[][] regionsToFlush =
-                wal.rollWriter(periodic || entry.getValue().booleanValue());
-            walNeedsRoll.put(wal, Boolean.FALSE);
+            final byte[][] regionsToFlush = wal.rollWriter(periodic ||
+                walInfo.getFirst().booleanValue() || syncFailed, syncFailed);
+            walInfo.setFirst(Boolean.FALSE);
             if (regionsToFlush != null) {
               for (byte[] r : regionsToFlush) {
                 scheduleFlush(r);
@@ -247,8 +255,8 @@ private void scheduleFlush(final byte [] encodedRegionName) {
    * @return true if all WAL roll finished
    */
   public boolean walRollFinished() {
-    for (boolean needRoll : walNeedsRoll.values()) {
-      if (needRoll) {
+    for (Pair<Boolean,Boolean> walInfo : walNeedsRoll.values()) {
+      if (walInfo.getFirst().booleanValue()) {
         return false;
       }
     }
@@ -271,7 +279,7 @@ public void close() {
   }
 
   @VisibleForTesting
-  Map<WAL, Boolean> getWalNeedsRoll() {
+  Map<WAL, Pair<Boolean,Boolean>> getWalNeedsRoll() {
     return this.walNeedsRoll;
   }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java

@@ -122,8 +122,8 @@ public abstract class AbstractFSWAL<W extends WriterBase> implements WAL {
   private static final String ROLL_ON_SYNC_TIME_MS = "hbase.regionserver.hlog.roll.on.sync.ms";
   protected static final int DEFAULT_ROLL_ON_SYNC_TIME_MS = 10000; // in ms
 
-  private static final String WAL_SYNC_TIMEOUT_MS = "hbase.regionserver.hlog.sync.timeout";
-  private static final int DEFAULT_WAL_SYNC_TIMEOUT_MS = 5 * 60 * 1000; // in ms, 5min
+  public static final String WAL_SYNC_TIMEOUT_MS = "hbase.regionserver.hlog.sync.timeout";
+  public static final int DEFAULT_WAL_SYNC_TIMEOUT_MS = 60 * 1000; // in ms, 5min
 
   /**
    * file system instance
@@ -516,7 +516,7 @@ public long getEarliestMemStoreSeqNum(byte[] encodedRegionName, byte[] familyNam
 
   @Override
   public byte[][] rollWriter() throws FailedLogCloseException, IOException {
-    return rollWriter(false);
+    return rollWriter(false, false);
   }
 
   /**
@@ -795,22 +795,25 @@ private IOException convertInterruptedExceptionToIOException(final InterruptedEx
   }
 
   @Override
-  public byte[][] rollWriter(boolean force) throws FailedLogCloseException, IOException {
+  public byte[][] rollWriter(boolean force, boolean syncFailed) throws FailedLogCloseException, IOException {
     rollWriterLock.lock();
     try {
       if (this.closed) {
         throw new WALClosedException("WAL has been closed");
       }
       // Return if nothing to flush.
-      if (!force && this.writer != null && this.numEntries.get() <= 0) {
+      if (!force && !syncFailed && (this.writer != null && this.numEntries.get() <= 0)) {
         return null;
       }
       byte[][] regionsToFlush = null;
       try (TraceScope scope = TraceUtil.createTrace("FSHLog.rollWriter")) {
         Path oldPath = getOldPath();
         Path newPath = getNewPath();
+        if (syncFailed){
+          LOG.info(oldPath + " rolled because of syncFailed!");
+        }
         // Any exception from here on is catastrophic, non-recoverable so we currently abort.
-        W nextWriter = this.createWriterInstance(newPath);
+        W nextWriter = this.createWriterInstance(newPath, syncFailed ? oldPath : null);
         tellListenersAboutPreLogRoll(oldPath, newPath);
         // NewPath could be equal to oldPath if replaceWriter fails.
         newPath = replaceWriter(oldPath, newPath, nextWriter);
@@ -843,11 +846,6 @@ public long getLogFileSize() {
     return this.totalLogSize.get();
   }
 
-  // public only until class moves to o.a.h.h.wal
-  public void requestLogRoll() {
-    requestLogRoll(false);
-  }
-
   /**
    * Get the backing files associated with this WAL.
    * @return may be null if there are no files.
@@ -925,10 +923,11 @@ protected final SyncFuture getSyncFuture(long sequence) {
     return cachedSyncFutures.get().reset(sequence);
   }
 
-  protected final void requestLogRoll(boolean tooFewReplicas) {
+   // public only until class moves to o.a.h.h.wal
+  public final void requestLogRoll(boolean tooFewReplicas, boolean syncFaild) {
     if (!this.listeners.isEmpty()) {
       for (WALActionsListener i : this.listeners) {
-        i.logRollRequested(tooFewReplicas);
+        i.logRollRequested(tooFewReplicas, syncFaild);
       }
     }
   }
@@ -1083,7 +1082,7 @@ public abstract long append(RegionInfo info, WALKeyImpl key, WALEdit edits, bool
 
   protected abstract void doAppend(W writer, FSWALEntry entry) throws IOException;
 
-  protected abstract W createWriterInstance(Path path)
+  protected abstract W createWriterInstance(Path path, Path oldPath)
       throws IOException, CommonFSUtils.StreamLacksCapabilityException;
 
   protected abstract void doReplaceWriter(Path oldPath, Path newPath, W nextWriter)
@@ -1105,7 +1104,7 @@ public void checkLogLowReplication(long checkInterval) {
     try {
       lastTimeCheckLowReplication = now;
       if (doCheckLogLowReplication()) {
-        requestLogRoll(true);
+        requestLogRoll(true, false);
       }
     } finally {
       rollWriterLock.unlock();

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractProtobufLogWriter.java

@@ -154,16 +154,16 @@ private boolean initializeCompressionContext(Configuration conf, Path path) thro
     return doCompress;
   }
 
-  public void init(FileSystem fs, Path path, Configuration conf, boolean overwritable,
-      long blocksize) throws IOException, StreamLacksCapabilityException {
+  public void init(FileSystem fs, Path path, Path oldPath, Configuration conf,
+      boolean overwritable, long blocksize) throws IOException, StreamLacksCapabilityException {
     this.conf = conf;
     boolean doCompress = initializeCompressionContext(conf, path);
     this.trailerWarnSize = conf.getInt(WAL_TRAILER_WARN_SIZE, DEFAULT_WAL_TRAILER_WARN_SIZE);
     int bufferSize = FSUtils.getDefaultBufferSize(fs);
     short replication = (short) conf.getInt("hbase.regionserver.hlog.replication",
       FSUtils.getDefaultReplication(fs, path));
 
-    initOutput(fs, path, overwritable, bufferSize, replication, blocksize);
+    initOutput(fs, path, oldPath, overwritable, bufferSize, replication, blocksize);
 
     boolean doTagCompress = doCompress
         && conf.getBoolean(CompressionContext.ENABLE_WAL_TAGS_COMPRESSION, true);
@@ -239,8 +239,9 @@ protected void writeWALTrailer() {
     }
   }
 
-  protected abstract void initOutput(FileSystem fs, Path path, boolean overwritable, int bufferSize,
-      short replication, long blockSize) throws IOException, StreamLacksCapabilityException;
+  protected abstract void initOutput(FileSystem fs, Path path, Path oldPath,
+      boolean overwritable, int bufferSize, short replication, long blockSize)
+          throws IOException, StreamLacksCapabilityException;
 
   /**
    * return the file length after written.