testWriteStateMachineDataFailure

Author: szetszwo

ratis-common/src/main/java/org/apache/ratis/util/LogUtils.java

@@ -126,7 +126,7 @@ static void warn(Logger log, Supplier<String> message, Throwable t, Class<?>...
         }
         log.warn(b.toString());
       } else {
-        log.warn(message.get(), t);
+        log.warn(message.get() + ": " + t);
       }
     }
   }

ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileInfo.java

@@ -276,7 +276,7 @@ CompletableFuture<Integer> submitCommit(
       final WriteInfo info = writeInfos.get(index);
       if (info == null) {
         return JavaUtils.completeExceptionally(
-            new IOException(name.get() + " is already committed."));
+            new IOException(name.get() + " not found."));
       }
 
       final CheckedSupplier<Integer, IOException> task = LogUtils.newCheckedSupplier(LOG, () -> {

ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreClient.java

@@ -171,10 +171,15 @@ private static <OUTPUT, THROWABLE extends Throwable> OUTPUT watchImpl(
 
   public long write(String path, long offset, boolean close, ByteBuffer buffer, boolean sync)
       throws IOException {
-    final int chunkSize = FileStoreCommon.getChunkSize(buffer.remaining());
-    buffer.limit(chunkSize);
-    final ByteString reply = writeImpl(this::send, path, offset, close, buffer, sync);
-    return WriteReplyProto.parseFrom(reply).getLength();
+    try {
+      final int chunkSize = FileStoreCommon.getChunkSize(buffer.remaining());
+      buffer.limit(chunkSize);
+      final ByteString reply = writeImpl(this::send, path, offset, close, buffer, sync);
+      return WriteReplyProto.parseFrom(reply).getLength();
+    } catch (IOException e) {
+      LOG.error("XXX Write failed for {} at offset {}, close? {}, sync? {}", path, offset, close, sync, e);
+      throw e;
+    }
   }
 
   public DataStreamOutput getStreamOutput(String path, long dataSize, RoutingTable routingTable) {

ratis-examples/src/main/java/org/apache/ratis/examples/filestore/FileStoreStateMachine.java

@@ -31,6 +31,7 @@
 import org.apache.ratis.protocol.Message;
 import org.apache.ratis.protocol.RaftClientRequest;
 import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.protocol.RaftPeerId;
 import org.apache.ratis.server.RaftServer;
 import org.apache.ratis.server.storage.RaftStorage;
 import org.apache.ratis.statemachine.StateMachineStorage;
@@ -44,6 +45,7 @@
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class FileStoreStateMachine extends BaseStateMachine {
   private final SimpleStateMachineStorage storage = new SimpleStateMachineStorage();
@@ -120,6 +122,8 @@ public TransactionContext startTransaction(LogEntryProto entry, RaftProtos.RaftP
         .build();
   }
 
+  static final AtomicReference<RaftPeerId> chosenFollower = new AtomicReference<>();
+
   @Override
   public CompletableFuture<Integer> write(LogEntryProto entry, TransactionContext context) {
     final FileStoreRequestProto proto = getProto(context, entry);
@@ -128,6 +132,27 @@ public CompletableFuture<Integer> write(LogEntryProto entry, TransactionContext
     }
 
     final WriteRequestHeaderProto h = proto.getWriteHeader();
+    final String path = h.getPath().toStringUtf8();
+    try {
+      final RaftServer.Division division = getServer().join().getDivision(getGroupId());
+      final RaftPeerId serverId = division.getId();
+      final boolean isFollower = path.equals("foo") && division.getInfo().isFollower();
+      if (isFollower) {
+        final boolean shouldFail;
+        if (chosenFollower.compareAndSet(null, serverId)) {
+          LOG.info("XXX follower {} is chosen", serverId);
+          shouldFail = true;
+        } else {
+          shouldFail = serverId.equals(chosenFollower.get());
+        }
+        if (shouldFail) {
+          return FileStoreCommon.completeExceptionally(entry.getIndex(), getId() + ": XXX: Failed follower for " + path);
+        }
+      }
+    } catch (IOException e) {
+      LOG.error("XXX Failed to check follower {}", path, e);
+    }
+
     final CompletableFuture<Integer> f = files.write(entry.getIndex(),
         h.getPath().toStringUtf8(), h.getClose(),  h.getSync(), h.getOffset(),
         entry.getStateMachineLogEntry().getStateMachineEntry().getStateMachineData());

ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreBaseTest.java

@@ -144,6 +144,24 @@ public void testFileStore() throws Exception {
     cluster.shutdown();
   }
 
+  @Test
+  public void testWriteStateMachineDataFailure() throws Exception {
+    final CLUSTER cluster = newCluster(NUM_PEERS);
+    cluster.start();
+    RaftTestUtil.waitForLeader(cluster);
+
+    final CheckedSupplier<FileStoreClient, IOException> newClient = () -> newFileStoreClient(cluster);
+
+    try {
+      testSingleFile("foo", SizeInBytes.valueOf("1m"), newClient);
+    } catch (Exception e) {
+      LOG.info("XXX Failed: foo", e);
+    }
+
+    testSingleFile("bar", SizeInBytes.valueOf("2k"), newClient);
+    cluster.shutdown();
+  }
+
   private static FileStoreWriter writeSingleFile(
       String path, SizeInBytes fileLength, CheckedSupplier<FileStoreClient, IOException> newClient)
       throws Exception {

ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreWriter.java

@@ -221,6 +221,8 @@ FileStoreWriter verify() throws IOException {
       verify(read, offset, n, expected);
       offset += n;
     }
+
+    LOG.info("XXX Verify successful: {}", fileName);
     return this;
   }
 

ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java

@@ -97,7 +97,7 @@ static Throwable unwrapThrowable(Throwable t) {
         return unwrapped;
       }
     }
-    return t;
+    return JavaUtils.unwrapCompletionException(t);
   }
 
   static IOException unwrapException(StatusRuntimeException se) {

ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java

@@ -61,6 +61,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -168,6 +169,7 @@ synchronized int process(Event event) {
   private final TimeoutExecutor scheduler = TimeoutExecutor.getInstance();
   @SuppressWarnings({"squid:S3077"}) // Suppress volatile for generic type
   private volatile StreamObservers appendLogRequestObserver;
+  private final AtomicBoolean alreadyReset = new AtomicBoolean(false);
   private final boolean useSeparateHBChannel;
 
   private final GrpcServerMetrics grpcServerMetrics;
@@ -212,6 +214,16 @@ private GrpcServerProtocolClient getClient() throws IOException {
   }
 
   private void resetClient(AppendEntriesRequest request, Event event) {
+    if (!alreadyReset.compareAndSet(false, true)) {
+      try {
+        final TimeDuration sleep = TimeDuration.ONE_SECOND;
+        LOG.warn("XXX {}: already reset client, sleep {}", this, sleep);
+        sleep.sleep();
+      } catch (InterruptedException e) {
+        throw new IllegalStateException(e);
+      }
+    }
+
     try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
       getClient().resetConnectBackoff();
       if (appendLogRequestObserver != null) {
@@ -511,6 +523,7 @@ private void onNextImpl(AppendEntriesRequest request, AppendEntriesReplyProto re
 
       switch (reply.getResult()) {
         case SUCCESS:
+          alreadyReset.set(false);
           grpcServerMetrics.onRequestSuccess(getFollowerId().toString(), reply.getIsHearbeat());
           getLeaderState().onFollowerCommitIndex(getFollower(), reply.getFollowerCommit());
           if (getFollower().updateMatchIndex(reply.getMatchIndex())) {
@@ -675,6 +688,7 @@ public void onNext(InstallSnapshotReplyProto reply) {
       final long followerSnapshotIndex;
       switch (reply.getResult()) {
         case SUCCESS:
+          alreadyReset.set(false);
           LOG.info("{}: Completed", this);
           getFollower().setAttemptedToInstallSnapshot();
           removePending(reply);

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -159,7 +159,7 @@ private void runImpl() {
         synchronized (server) {
           if (roleChangeChecking(electionTimeout)) {
             LOG.info("{}: change to CANDIDATE, lastRpcElapsedTime:{}, electionTimeout:{}",
-                this, lastRpcTime.elapsedTime(), electionTimeout);
+                this, lastRpcTime.elapsedTime().toString(TimeUnit.MILLISECONDS, 3), electionTimeout);
             server.getLeaderElectionMetrics().onLeaderElectionTimeout(); // Update timeout metric counters.
             // election timeout, should become a candidate
             server.changeToCandidate(false);

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -1698,6 +1698,12 @@ leaderId, getMemberId(), currentTerm, followerCommit, inconsistencyReplyNextInde
     final long commitIndex = effectiveCommitIndex(proto.getLeaderCommit(), previous, entries.size());
     final long matchIndex = isHeartbeat? RaftLog.INVALID_LOG_INDEX: entries.get(entries.size() - 1).getIndex();
     return appendFuture.whenCompleteAsync((r, t) -> {
+      if  (t != null) {
+        LOG.error("{}: XXX appendEntries* failed: {}", getMemberId(), t.toString());
+      } else if (!entries.isEmpty()) {
+        LOG.info("{}: XXX appendEntries* succeeded {} entries: {} -> {}",
+            getMemberId(), entries.size(), entries.get(0).getIndex(), entries.get(entries.size() - 1).getIndex());
+      }
       followerState.ifPresent(fs -> fs.updateLastRpcTime(FollowerState.UpdateType.APPEND_COMPLETE));
       timer.stop();
     }, getServerExecutor()).thenApply(v -> {