apache · cbaenziger · Nov 5, 2019
diff --git a/dev-support/vagrant/namazu_configs/hdd_config.toml b/dev-support/vagrant/namazu_configs/hdd_config.toml
@@ -23,4 +23,4 @@ explorePolicy = "random"
   maxInterval = "300ms"
 
   # for Filesystem inspectors, you can specify fault-injection probability (0.0-1.0).
-  faultActionProbability = 0.0
+  faultActionProbability = 0.4
diff --git a/pom.xml b/pom.xml
@@ -408,6 +408,12 @@
         <version>${dropwizard.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>net.jodah</groupId>
+        <artifactId>failsafe</artifactId>
+        <version>2.3.1</version>
+      </dependency>
+
       <dependency>
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>

diff --git a/ratis-common/pom.xml b/ratis-common/pom.xml
@@ -42,6 +42,11 @@
       <artifactId>slf4j-log4j12</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>net.jodah</groupId>
+      <artifactId>failsafe</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

diff --git a/ratis-common/src/main/java/org/apache/ratis/retry/IORetryPolicy.java b/ratis-common/src/main/java/org/apache/ratis/retry/IORetryPolicy.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ratis.retry;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import net.jodah.failsafe.RetryPolicy;
+import java.time.Duration;
+import java.io.IOException;
+
+/*
+ * An interface to hold various I/O retry policies
+ */
+public interface IORetryPolicy {
+  Logger LOG = LoggerFactory.getLogger(IORetryPolicy.class);
+
+  public final RetryPolicy<Object> retryPolicy = new RetryPolicy<Object>()
+      .handle(IOException.class)
+      .onRetry(e -> LOG.warn("Retrying:", (e.getLastFailure() != null ? e.getLastFailure() :e.getLastResult())))
+      .onFailure(e -> LOG.error("Failed:", e.getFailure()))
+      .withDelay(Duration.ofNanos(1))
+      .withMaxAttempts(-1);
+
+  public final RetryPolicy<Boolean> booleanCheckingRetryPolicy = new RetryPolicy<Boolean>()
+      .handle(IOException.class)
+      .handleResult(false)
+      .onRetry(e -> LOG.warn("Retrying boolean:", e.getLastFailure()))
+      .onFailure(e -> LOG.error("Failed to get true for:", e.getFailure()))
+      .withDelay(Duration.ofNanos(1))
+      .withMaxAttempts(-1);
+} 
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/AtomicFileOutputStream.java b/ratis-common/src/main/java/org/apache/ratis/util/AtomicFileOutputStream.java
@@ -20,8 +20,15 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.FailsafeException;
+import net.jodah.failsafe.function.CheckedRunnable;
+import net.jodah.failsafe.function.CheckedSupplier;
+
 import java.io.*;
 
+import org.apache.ratis.retry.IORetryPolicy;
+
 /**
  * A FileOutputStream that has the property that it will only show
  * up at its destination once it has been entirely written and flushed
@@ -56,29 +63,41 @@ public AtomicFileOutputStream(File f) throws FileNotFoundException {
   public void close() throws IOException {
     boolean triedToClose = false, success = false;
     try {
-      flush();
-      ((FileOutputStream)out).getChannel().force(true);
-
+      // WARNING: We try a LOT more than 5 times for this function as each step is
+      // retried independently to keep the original code flow
+      Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+        flush();
+        ((FileOutputStream)out).getChannel().force(true);
+      });
       triedToClose = true;
-      super.close();
+      Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+        super.close();
+      });
       success = true;
     } finally {
       if (success) {
-        boolean renamed = tmpFile.renameTo(origFile);
+        boolean renamed = Failsafe.with(IORetryPolicy.booleanCheckingRetryPolicy).get((CheckedSupplier<Boolean>)()->{
+          return(tmpFile.renameTo(origFile));
+        });
         if (!renamed) {
           // On windows, renameTo does not replace.
-          if (origFile.exists() && !origFile.delete()) {
+          boolean exists = Failsafe.with(IORetryPolicy.booleanCheckingRetryPolicy).get((CheckedSupplier<Boolean>)(origFile::exists));
+          if (exists && !Failsafe.with(IORetryPolicy.booleanCheckingRetryPolicy).get((CheckedSupplier<Boolean>)(origFile::delete))) {
             throw new IOException("Could not delete original file " + origFile);
           }
-          FileUtils.move(tmpFile, origFile);
+          Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+            FileUtils.move(tmpFile, origFile);
+          });
         }
       } else {
         if (!triedToClose) {
           // If we failed when flushing, try to close it to not leak an FD
-          IOUtils.cleanup(LOG, out);
+          Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+            IOUtils.cleanup(LOG, out);
+          });
         }
         // close wasn't successful, try to delete the tmp file
-        if (!tmpFile.delete()) {
+        if (!Failsafe.with(IORetryPolicy.booleanCheckingRetryPolicy).get((CheckedSupplier<Boolean>)(tmpFile::delete))) {
           LOG.warn("Unable to delete tmp file " + tmpFile);
         }
       }
@@ -92,11 +111,11 @@ public void close() throws IOException {
    */
   public void abort() {
     try {
-      super.close();
-    } catch (IOException ioe) {
-      LOG.warn("Unable to abort file " + tmpFile, ioe);
+      Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)(super::close));
+    } catch (FailsafeException e) {
+      LOG.warn("Unable to abort file " + tmpFile, e);
     }
-    if (!tmpFile.delete()) {
+    if (!Failsafe.with(IORetryPolicy.booleanCheckingRetryPolicy).get((CheckedSupplier<Boolean>)(tmpFile::delete))) {
       LOG.warn("Unable to delete tmp file during abort " + tmpFile);
     }
   }

diff --git a/ratis-common/src/main/java/org/apache/ratis/util/FileUtils.java b/ratis-common/src/main/java/org/apache/ratis/util/FileUtils.java
@@ -17,7 +17,6 @@
  */
 package org.apache.ratis.util;
 
-import org.apache.ratis.util.function.CheckedSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -27,31 +26,24 @@
 import java.io.OutputStream;
 import java.nio.file.*;
 import java.nio.file.attribute.BasicFileAttributes;
-import java.util.function.Supplier;
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.function.CheckedRunnable;
+
+import org.apache.ratis.retry.IORetryPolicy;
 
 public interface FileUtils {
   Logger LOG = LoggerFactory.getLogger(FileUtils.class);
 
-  int NUM_ATTEMPTS = 5;
-  TimeDuration SLEEP_TIME = TimeDuration.ONE_SECOND;
-
-  static <T> T attempt(CheckedSupplier<T, IOException> op, Supplier<?> name) throws IOException {
-    try {
-      return JavaUtils.attempt(op, NUM_ATTEMPTS, SLEEP_TIME, name, LOG);
-    } catch (InterruptedException e) {
-      throw IOUtils.toInterruptedIOException("Interrupted " + name.get(), e);
-    }
-  }
-
   static void truncateFile(File f, long target) throws IOException {
-    final long original = f.length();
-    LogUtils.runAndLog(LOG,
-        () -> {
-          try (FileOutputStream out = new FileOutputStream(f, true)) {
-            out.getChannel().truncate(target);
-          }
-        },
-        () -> "FileOutputStream.getChannel().truncate " + f + " length: " + original + " -> " + target);
+    Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+      final long original = f.length();
+      LogUtils.runAndLog(LOG,
+          () -> {
+            try (FileOutputStream out = new FileOutputStream(f, true)) {
+              out.getChannel().truncate(target);
+            }},
+          () -> "FileOutputStream.getChannel().truncate " + f + " length: " + original + " -> " + target);
+    });
   }
 
   static OutputStream createNewFile(Path p) throws IOException {
@@ -65,19 +57,23 @@ static void createDirectories(File dir) throws IOException {
   }
 
   static void createDirectories(Path dir) throws IOException {
-    LogUtils.runAndLog(LOG,
-        () -> Files.createDirectories(dir),
-        () -> "Files.createDirectories " + dir);
+    Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+      LogUtils.runAndLog(LOG,
+          () -> Files.createDirectories(dir),
+          () -> "Files.createDirectories " + dir);
+    });
   }
 
   static void move(File src, File dst) throws IOException {
     move(src.toPath(), dst.toPath());
   }
 
   static void move(Path src, Path dst) throws IOException {
-    LogUtils.runAndLog(LOG,
-        () -> Files.move(src, dst),
-        () -> "Files.move " + src + " to " + dst);
+    Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+      LogUtils.runAndLog(LOG,
+          () -> Files.move(src, dst),
+          () -> "Files.move " + src + " to " + dst);
+    });
   }
 
   /** The same as passing f.toPath() to {@link #delete(Path)}. */
@@ -100,9 +96,11 @@ static void deleteFileQuietly(File f) {
    * This method may print log messages using {@link #LOG}.
    */
   static void delete(Path p) throws IOException {
-    LogUtils.runAndLog(LOG,
-        () -> Files.delete(p),
-        () -> "Files.delete " + p);
+    Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+      LogUtils.runAndLog(LOG,
+          () -> Files.delete(p),
+          () -> "Files.delete " + p);
+    });
   }
 
   /** The same as passing f.toPath() to {@link #deleteFully(Path)}. */
@@ -122,26 +120,28 @@ static void deleteFully(File f) throws IOException {
    * (3) If it is a symlink, the symlink will be deleted but the symlink target will not be deleted.
    */
   static void deleteFully(Path p) throws IOException {
-    if (!Files.exists(p, LinkOption.NOFOLLOW_LINKS)) {
-      LOG.trace("deleteFully: {} does not exist.", p);
-      return;
-    }
-    Files.walkFileTree(p, new SimpleFileVisitor<Path>() {
-      @Override
-      public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
-        delete(file);
-        return FileVisitResult.CONTINUE;
+    Failsafe.with(IORetryPolicy.retryPolicy).run((CheckedRunnable)()->{
+      if (!Files.exists(p, LinkOption.NOFOLLOW_LINKS)) {
+        LOG.trace("deleteFully: {} does not exist.", p);
+        return;
       }
+      Files.walkFileTree(p, new SimpleFileVisitor<Path>() {
+        @Override
+        public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+          delete(file);
+          return FileVisitResult.CONTINUE;
+        }
 
-      @Override
-      public FileVisitResult postVisitDirectory(Path dir, IOException e) throws IOException {
-        if (e != null) {
-          // directory iteration failed
-          throw e;
+        @Override
+        public FileVisitResult postVisitDirectory(Path dir, IOException e) throws IOException {
+          if (e != null) {
+            // directory iteration failed
+            throw e;
+          }
+          delete(dir);
+          return FileVisitResult.CONTINUE;
         }
-        delete(dir);
-        return FileVisitResult.CONTINUE;
-      }
+      });
     });
   }
 }
diff --git a/ratis-server/pom.xml b/ratis-server/pom.xml
@@ -79,6 +79,10 @@
       <artifactId>mockito-all</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>net.jodah</groupId>
+      <artifactId>failsafe</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.ratis</groupId>
       <artifactId>ratis-metrics</artifactId>

diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LogAppender.java
@@ -29,6 +29,12 @@
 import org.apache.ratis.proto.RaftProtos.*;
 import org.apache.ratis.statemachine.SnapshotInfo;
 import org.apache.ratis.util.*;
+import org.apache.ratis.retry.IORetryPolicy;
+
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.function.CheckedRunnable;
+import net.jodah.failsafe.function.CheckedSupplier;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -346,8 +352,11 @@ public InstallSnapshotRequestProto next() {
               currentFileSize - currentOffset);
           FileChunkProto chunk;
           try {
-            chunk = readFileChunk(currentFileInfo, in, currentBuf,
-                targetLength, currentOffset, chunkIndex);
+            chunk = Failsafe.with(IORetryPolicy.retryPolicy).get((
+                CheckedSupplier<FileChunkProto>)()->{
+                return(readFileChunk(currentFileInfo, in, currentBuf,
+                    targetLength, currentOffset, chunkIndex));
+                });
             boolean done = (fileIndex == files.size() - 1) &&
                 chunk.getDone();
             InstallSnapshotRequestProto request =
@@ -391,7 +400,10 @@ private FileChunkProto readFileChunk(FileInfo fileInfo,
       throws IOException {
     FileChunkProto.Builder builder = FileChunkProto.newBuilder()
         .setOffset(offset).setChunkIndex(chunkIndex);
-    IOUtils.readFully(in, buf, 0, length);
+    Failsafe.with(IORetryPolicy.retryPolicy).run((
+        CheckedRunnable)()->{
+          IOUtils.readFully(in, buf, 0, length);
+        });
     Path relativePath = server.getState().getStorage().getStorageDir()
         .relativizeToRoot(fileInfo.getPath());
     builder.setFilename(relativePath.toString());