RATIS-2345. Leader stepDown could cause a deadlock. (#1300)

Author: szetszwo

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

@@ -703,10 +703,12 @@ void submitStepDownEvent(long term, StepDownReason reason) {
   private void stepDown(long term, StepDownReason reason) {
     try {
       lease.getAndSetEnabled(false);
-      server.changeToFollowerAndPersistMetadata(term, false, reason).join();
+      server.changeToFollowerAndPersistMetadata(term, false, reason)
+          .get(5, TimeUnit.SECONDS);
       pendingStepDown.complete(server::newSuccessReply);
-    } catch(IOException e) {
-      final String s = this + ": Failed to persist metadata for term " + term;
+    } catch(Exception e) {
+      pendingStepDown.completeExceptionally(e);
+      final String s = this + ": Failed to step down for term " + term;
       LOG.warn(s, e);
       // the failure should happen while changing the state to follower
       // thus the in-memory state should have been updated

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

@@ -56,8 +56,12 @@ void complete(Function<TransferLeadershipRequest, RaftClientReply> newSuccessRep
       replyFuture.complete(newSuccessReply.apply(request));
     }
 
+    void completeExceptionally(Exception e) {
+      replyFuture.completeExceptionally(e);
+    }
+
     void timeout() {
-      replyFuture.completeExceptionally(new TimeoutIOException(
+      completeExceptionally(new TimeoutIOException(
           ": Failed to step down leader on " +  leader + "request " + request.getTimeoutMs() + "ms"));
     }
 
@@ -105,6 +109,10 @@ void complete(Function<TransferLeadershipRequest, RaftClientReply> newSuccessRep
     pending.getAndSetNull().ifPresent(p -> p.complete(newSuccessReply));
   }
 
+  void completeExceptionally(Exception e) {
+    pending.getAndSetNull().ifPresent(p -> p.completeExceptionally(e));
+  }
+
   void timeout() {
     pending.getAndSetNull().ifPresent(PendingRequest::timeout);
   }

ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java

@@ -360,7 +360,7 @@ static void testWithLoad(final int numClients, final int numMessages,
 
     final Timer timer = new Timer();
     timer.schedule(new TimerTask() {
-      private int previousLastStep = lastStep.get();
+      private final AtomicInteger previousLastStep = new AtomicInteger(lastStep.get());
 
       @Override
       public void run() {
@@ -371,8 +371,8 @@ public void run() {
         JavaUtils.dumpAllThreads(s -> log.info(s));
 
         final int last = lastStep.get();
-        if (last != previousLastStep) {
-          previousLastStep = last;
+        if (last != previousLastStep.get()) {
+          previousLastStep.set(last);
         } else {
           final RaftServer.Division leader = cluster.getLeader();
           log.info("NO PROGRESS at " + last + ", try to restart leader=" + leader);