RATIS-2271 Leadership Loss Causes ClosedByInterruptException and NullPointerException in LogAppender Thread (#1245)

Author: RkGrit

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogSegment.java

@@ -292,7 +292,11 @@ ReferenceCountedObject<LogEntryProto> load(TermIndex key) throws IOException {
         }
       });
       loadingTimes.incrementAndGet();
-      return Objects.requireNonNull(toReturn.get(), () -> "toReturn == null for " + key);
+      final ReferenceCountedObject<LogEntryProto> proto = toReturn.get();
+      if (proto == null) {
+        throw new RaftLogIOException("Failed to load log entry " + key);
+      }
+      return proto;
     }
   }
 
@@ -502,8 +506,10 @@ synchronized ReferenceCountedObject<LogEntryProto> loadCache(TermIndex ti) throw
     }
     try {
       return cacheLoader.load(ti);
+    } catch (RaftLogIOException e) {
+      throw e;
     } catch (Exception e) {
-      throw new RaftLogIOException(e);
+      throw new RaftLogIOException("Failed to loadCache for log entry " + ti, e);
     }
   }
 

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogInputStream.java

@@ -21,6 +21,7 @@
 import java.io.EOFException;
 import java.io.File;
 import java.io.IOException;
+import java.nio.channels.ClosedByInterruptException;
 import java.util.Optional;
 
 import org.apache.ratis.proto.RaftProtos.LogEntryProto;
@@ -104,7 +105,11 @@ public LogEntryProto nextEntry() throws IOException {
         try {
           init();
         } catch (Exception e) {
-          LOG.error("caught exception initializing " + this, e);
+          if (e.getCause() instanceof ClosedByInterruptException) {
+            LOG.warn("Initialization is interrupted: {}", this, e);
+          } else {
+            LOG.error("Failed to initialize {}", this, e);
+          }
           throw IOUtils.asIOException(e);
         }
     }

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java

@@ -42,6 +42,7 @@
 import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.channels.ClosedByInterruptException;
 import java.util.Optional;
 import java.util.zip.Checksum;
 
@@ -169,7 +170,14 @@ public long skip(long amt) throws IOException {
    */
   boolean verifyHeader() throws IOException {
     final int headerLength = SegmentedRaftLogFormat.getHeaderLength();
-    final int readLength = in.read(temp, 0, headerLength);
+    final int readLength;
+    try{
+      readLength = in.read(temp, 0, headerLength);
+    } catch (ClosedByInterruptException e) {
+      Thread.currentThread().interrupt();
+      throw new IOException("Interrupted while reading the header of " + file, e);
+    }
+
     Preconditions.assertTrue(readLength <= headerLength);
     final int matchLength = SegmentedRaftLogFormat.matchHeader(temp, 0, readLength);
     Preconditions.assertTrue(matchLength <= readLength);