diff --git a/ratis-client/src/main/java/org/apache/ratis/client/api/AsyncApi.java b/ratis-client/src/main/java/org/apache/ratis/client/api/AsyncApi.java
index 483a222..e383281 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/api/AsyncApi.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/api/AsyncApi.java
@@ -36,9 +36,15 @@
    * For readonly messages, use {@link #sendReadOnly(Message)} instead.
    *
    * @param message The request message.
+   * @param replication The replication level to wait for.
    * @return a future of the reply.
    */
-  CompletableFuture<RaftClientReply> send(Message message);
+  CompletableFuture<RaftClientReply> send(Message message, ReplicationLevel replication);
+
+  /** The same as send(message, ReplicationLevel.MAJORITY). */
+  default CompletableFuture<RaftClientReply> send(Message message) {
+    return send(message, ReplicationLevel.MAJORITY);
+  }
 
   /** The same as sendReadOnly(message, null). */
   default CompletableFuture<RaftClientReply> sendReadOnly(Message message) {
diff --git a/ratis-client/src/main/java/org/apache/ratis/client/impl/AsyncImpl.java b/ratis-client/src/main/java/org/apache/ratis/client/impl/AsyncImpl.java
index 2f7069f..8547ce2 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/impl/AsyncImpl.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/impl/AsyncImpl.java
@@ -42,8 +42,8 @@
   }
 
   @Override
-  public CompletableFuture<RaftClientReply> send(Message message) {
-    return send(RaftClientRequest.writeRequestType(), message, null);
+  public CompletableFuture<RaftClientReply> send(Message message, ReplicationLevel replication) {
+    return send(RaftClientRequest.writeRequestType(replication), message, null);
   }
 
   @Override
diff --git a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
index 9d853b4..ed41f1e 100644
--- a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
+++ b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
@@ -17,23 +17,31 @@
  */
 package org.apache.ratis.protocol;
 
-import org.apache.ratis.proto.RaftProtos.*;
+import org.apache.ratis.proto.RaftProtos.DataStreamRequestTypeProto;
+import org.apache.ratis.proto.RaftProtos.ForwardRequestTypeProto;
+import org.apache.ratis.proto.RaftProtos.MessageStreamRequestTypeProto;
+import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto.TypeCase;
+import org.apache.ratis.proto.RaftProtos.ReadRequestTypeProto;
+import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
+import org.apache.ratis.proto.RaftProtos.SlidingWindowEntry;
+import org.apache.ratis.proto.RaftProtos.StaleReadRequestTypeProto;
+import org.apache.ratis.proto.RaftProtos.WatchRequestTypeProto;
+import org.apache.ratis.proto.RaftProtos.WriteRequestTypeProto;
 import org.apache.ratis.util.Preconditions;
 import org.apache.ratis.util.ProtoUtils;
 
 import java.util.Collections;
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 
-import static org.apache.ratis.proto.RaftProtos.RaftClientRequestProto.TypeCase.*;
-
 /**
  * Request from client to server
  */
 public class RaftClientRequest extends RaftClientMessage {
   private static final Type DATA_STREAM_DEFAULT = new Type(DataStreamRequestTypeProto.getDefaultInstance());
   private static final Type FORWARD_DEFAULT = new Type(ForwardRequestTypeProto.getDefaultInstance());
-  private static final Type WRITE_DEFAULT = new Type(WriteRequestTypeProto.getDefaultInstance());
   private static final Type WATCH_DEFAULT = new Type(
       WatchRequestTypeProto.newBuilder().setIndex(0L).setReplication(ReplicationLevel.MAJORITY).build());
 
@@ -44,8 +52,26 @@
       = new Type(ReadRequestTypeProto.newBuilder().setPreferNonLinearizable(true).build());
   private static final Type STALE_READ_DEFAULT = new Type(StaleReadRequestTypeProto.getDefaultInstance());
 
+  private static final Map<ReplicationLevel, Type> WRITE_REQUEST_TYPES;
+
+  static {
+    final EnumMap<ReplicationLevel, Type> map = new EnumMap<>(ReplicationLevel.class);
+    for(ReplicationLevel replication : ReplicationLevel.values()) {
+      if (replication == ReplicationLevel.UNRECOGNIZED) {
+        continue;
+      }
+      final WriteRequestTypeProto write = WriteRequestTypeProto.newBuilder().setReplication(replication).build();
+      map.put(replication, new Type(write));
+    }
+    WRITE_REQUEST_TYPES = Collections.unmodifiableMap(map);
+  }
+
+  public static Type writeRequestType(ReplicationLevel replication) {
+    return WRITE_REQUEST_TYPES.get(replication);
+  }
+
   public static Type writeRequestType() {
-    return WRITE_DEFAULT;
+    return writeRequestType(ReplicationLevel.MAJORITY);
   }
 
   public static Type dataStreamRequestType() {
@@ -88,10 +114,10 @@
     return new Type(WatchRequestTypeProto.newBuilder().setIndex(index).setReplication(replication).build());
   }
 
-  /** The type of {@link RaftClientRequest} corresponding to {@link RaftClientRequestProto.TypeCase}. */
+  /** The type of {@link RaftClientRequest} corresponding to {@link TypeCase}. */
   public static final class Type {
     public static Type valueOf(WriteRequestTypeProto write) {
-      return WRITE_DEFAULT;
+      return writeRequestType(write.getReplication());
     }
 
     public static Type valueOf(DataStreamRequestTypeProto dataStream) {
@@ -126,43 +152,43 @@
      * Only the corresponding proto (must be non-null) is used.
      * The other protos are ignored.
      */
-    private final RaftClientRequestProto.TypeCase typeCase;
+    private final TypeCase typeCase;
     private final Object proto;
 
-    private Type(RaftClientRequestProto.TypeCase typeCase, Object proto) {
+    private Type(TypeCase typeCase, Object proto) {
       this.typeCase = Objects.requireNonNull(typeCase, "typeCase == null");
       this.proto = Objects.requireNonNull(proto, "proto == null");
     }
 
     private Type(WriteRequestTypeProto write) {
-      this(WRITE, write);
+      this(TypeCase.WRITE, write);
     }
 
     private Type(DataStreamRequestTypeProto dataStream) {
-      this(DATASTREAM, dataStream);
+      this(TypeCase.DATASTREAM, dataStream);
     }
 
     private Type(ForwardRequestTypeProto forward) {
-      this(FORWARD, forward);
+      this(TypeCase.FORWARD, forward);
     }
 
     private Type(MessageStreamRequestTypeProto messageStream) {
-      this(MESSAGESTREAM, messageStream);
+      this(TypeCase.MESSAGESTREAM, messageStream);
     }
 
     private Type(ReadRequestTypeProto read) {
-      this(READ, read);
+      this(TypeCase.READ, read);
     }
 
     private Type(StaleReadRequestTypeProto staleRead) {
-      this(STALEREAD, staleRead);
+      this(TypeCase.STALEREAD, staleRead);
     }
 
     private Type(WatchRequestTypeProto watch) {
-      this(WATCH, watch);
+      this(TypeCase.WATCH, watch);
     }
 
-    public boolean is(RaftClientRequestProto.TypeCase t) {
+    public boolean is(TypeCase t) {
       return getTypeCase() == t;
     }
 
@@ -182,42 +208,46 @@
       }
     }
 
-    public RaftClientRequestProto.TypeCase getTypeCase() {
+    public TypeCase getTypeCase() {
       return typeCase;
     }
 
+    private void assertType(TypeCase expected) {
+      Preconditions.assertSame(expected, getTypeCase(), "type");
+    }
+
     public WriteRequestTypeProto getWrite() {
-      Preconditions.assertTrue(is(WRITE));
+      assertType(TypeCase.WRITE);
       return (WriteRequestTypeProto)proto;
     }
 
     public DataStreamRequestTypeProto getDataStream() {
-      Preconditions.assertTrue(is(DATASTREAM));
+      assertType(TypeCase.DATASTREAM);
       return (DataStreamRequestTypeProto)proto;
     }
 
     public ForwardRequestTypeProto getForward() {
-      Preconditions.assertTrue(is(FORWARD));
+      assertType(TypeCase.FORWARD);
       return (ForwardRequestTypeProto)proto;
     }
 
     public MessageStreamRequestTypeProto getMessageStream() {
-      Preconditions.assertTrue(is(MESSAGESTREAM), () -> "proto = " + proto);
+      assertType(TypeCase.MESSAGESTREAM);
       return (MessageStreamRequestTypeProto)proto;
     }
 
     public ReadRequestTypeProto getRead() {
-      Preconditions.assertTrue(is(READ));
+      assertType(TypeCase.READ);
       return (ReadRequestTypeProto)proto;
     }
 
     public StaleReadRequestTypeProto getStaleRead() {
-      Preconditions.assertTrue(is(STALEREAD));
+      assertType(TypeCase.STALEREAD);
       return (StaleReadRequestTypeProto)proto;
     }
 
     public WatchRequestTypeProto getWatch() {
-      Preconditions.assertTrue(is(WATCH));
+      assertType(TypeCase.WATCH);
       return (WatchRequestTypeProto)proto;
     }
 
@@ -426,7 +456,7 @@
     return type;
   }
 
-  public boolean is(RaftClientRequestProto.TypeCase typeCase) {
+  public boolean is(TypeCase typeCase) {
     return getType().is(typeCase);
   }
 
diff --git a/ratis-proto/src/main/proto/Raft.proto b/ratis-proto/src/main/proto/Raft.proto
index d8a1b62..586ec1b 100644
--- a/ratis-proto/src/main/proto/Raft.proto
+++ b/ratis-proto/src/main/proto/Raft.proto
@@ -281,6 +281,7 @@
 }
 
 message WriteRequestTypeProto {
+  ReplicationLevel replication = 1;
 }
 
 message MessageStreamRequestTypeProto {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
index c6983e3..b278891 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
@@ -545,7 +545,7 @@
         .thenApply(bytes -> RaftClientRequest.toWriteRequest(request, Message.valueOf(bytes)));
   }
 
-  CompletableFuture<RaftClientReply> addWatchReqeust(RaftClientRequest request) {
+  CompletableFuture<RaftClientReply> addWatchRequest(RaftClientRequest request) {
     LOG.debug("{}: addWatchRequest {}", this, request);
     return watchRequests.add(request)
         .thenApply(logIndex -> server.newSuccessReply(request, logIndex))
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
index 73451bf..51067a8 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
@@ -46,29 +46,8 @@
 import org.apache.ratis.client.impl.ClientProtoUtils;
 import org.apache.ratis.conf.RaftProperties;
 import org.apache.ratis.metrics.Timekeeper;
-import org.apache.ratis.proto.RaftProtos.AppendEntriesReplyProto;
-import org.apache.ratis.proto.RaftProtos.AppendEntriesRequestProto;
-import org.apache.ratis.proto.RaftProtos.CandidateInfoProto;
-import org.apache.ratis.proto.RaftProtos.CommitInfoProto;
-import org.apache.ratis.proto.RaftProtos.FollowerInfoProto;
-import org.apache.ratis.proto.RaftProtos.InstallSnapshotReplyProto;
-import org.apache.ratis.proto.RaftProtos.InstallSnapshotRequestProto;
-import org.apache.ratis.proto.RaftProtos.InstallSnapshotResult;
-import org.apache.ratis.proto.RaftProtos.LeaderInfoProto;
-import org.apache.ratis.proto.RaftProtos.LogEntryProto;
-import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto;
+import org.apache.ratis.proto.RaftProtos.*;
 import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto.TypeCase;
-import org.apache.ratis.proto.RaftProtos.RaftConfigurationProto;
-import org.apache.ratis.proto.RaftProtos.RaftPeerRole;
-import org.apache.ratis.proto.RaftProtos.RaftRpcRequestProto;
-import org.apache.ratis.proto.RaftProtos.ReadIndexReplyProto;
-import org.apache.ratis.proto.RaftProtos.ReadIndexRequestProto;
-import org.apache.ratis.proto.RaftProtos.RequestVoteReplyProto;
-import org.apache.ratis.proto.RaftProtos.RequestVoteRequestProto;
-import org.apache.ratis.proto.RaftProtos.RoleInfoProto;
-import org.apache.ratis.proto.RaftProtos.ServerRpcProto;
-import org.apache.ratis.proto.RaftProtos.StartLeaderElectionReplyProto;
-import org.apache.ratis.proto.RaftProtos.StartLeaderElectionRequestProto;
 import org.apache.ratis.protocol.ClientInvocationId;
 import org.apache.ratis.protocol.GroupInfoReply;
 import org.apache.ratis.protocol.GroupInfoRequest;
@@ -86,20 +65,7 @@
 import org.apache.ratis.protocol.SetConfigurationRequest;
 import org.apache.ratis.protocol.SnapshotManagementRequest;
 import org.apache.ratis.protocol.TransferLeadershipRequest;
-import org.apache.ratis.protocol.exceptions.GroupMismatchException;
-import org.apache.ratis.protocol.exceptions.LeaderNotReadyException;
-import org.apache.ratis.protocol.exceptions.LeaderSteppingDownException;
-import org.apache.ratis.protocol.exceptions.NotLeaderException;
-import org.apache.ratis.protocol.exceptions.RaftException;
-import org.apache.ratis.protocol.exceptions.ReadException;
-import org.apache.ratis.protocol.exceptions.ReadIndexException;
-import org.apache.ratis.protocol.exceptions.ReconfigurationInProgressException;
-import org.apache.ratis.protocol.exceptions.ResourceUnavailableException;
-import org.apache.ratis.protocol.exceptions.ServerNotReadyException;
-import org.apache.ratis.protocol.exceptions.SetConfigurationException;
-import org.apache.ratis.protocol.exceptions.StaleReadException;
-import org.apache.ratis.protocol.exceptions.StateMachineException;
-import org.apache.ratis.protocol.exceptions.TransferLeadershipException;
+import org.apache.ratis.protocol.exceptions.*;
 import org.apache.ratis.server.DataStreamMap;
 import org.apache.ratis.server.DivisionInfo;
 import org.apache.ratis.server.DivisionProperties;
@@ -597,7 +563,17 @@
     if ((old != RaftPeerRole.FOLLOWER || force) && old != RaftPeerRole.LISTENER) {
       setRole(RaftPeerRole.FOLLOWER, reason);
       if (old == RaftPeerRole.LEADER) {
-        role.shutdownLeaderState(false);
+        role.shutdownLeaderState(false)
+            .exceptionally(e -> {
+              if (e != null) {
+                if (!getInfo().isAlive()) {
+                  LOG.info("Since server is not alive {}, safely ignore {}", this, e.toString());
+                  return null;
+                }
+              }
+              throw new CompletionException("Failed to shutdownLeaderState: " + this, e);
+            })
+            .join();
         state.setLeader(null, reason);
       } else if (old == RaftPeerRole.CANDIDATE) {
         role.shutdownLeaderElection();
@@ -883,7 +859,30 @@
       }
       leaderState.notifySenders();
     }
-    return pending.getFuture();
+
+    final CompletableFuture<RaftClientReply> future = pending.getFuture();
+    if (request.is(TypeCase.WRITE)) {
+      // check replication
+      final ReplicationLevel replication = request.getType().getWrite().getReplication();
+      if (replication != ReplicationLevel.MAJORITY) {
+        return future.thenCompose(reply -> waitForReplication(reply, replication));
+      }
+    }
+
+    return future;
+  }
+
+  /** Wait until the given replication requirement is satisfied. */
+  private CompletableFuture<RaftClientReply> waitForReplication(RaftClientReply reply, ReplicationLevel replication) {
+    final RaftClientRequest.Type type = RaftClientRequest.watchRequestType(reply.getLogIndex(), replication);
+    final RaftClientRequest watch = RaftClientRequest.newBuilder()
+        .setServerId(reply.getServerId())
+        .setClientId(reply.getClientId())
+        .setGroupId(reply.getRaftGroupId())
+        .setCallId(reply.getCallId())
+        .setType(type)
+        .build();
+    return watchAsync(watch).thenApply(r -> reply);
   }
 
   void stepDownOnJvmPause() {
@@ -1002,7 +1001,7 @@
     }
 
     return role.getLeaderState()
-        .map(ls -> ls.addWatchReqeust(request))
+        .map(ls -> ls.addWatchRequest(request))
         .orElseGet(() -> CompletableFuture.completedFuture(
             newExceptionReply(request, generateNotLeaderException())));
   }
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RoleInfo.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RoleInfo.java
index 9f53b69..fe2bc96 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RoleInfo.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RoleInfo.java
@@ -20,6 +20,7 @@
 
 import org.apache.ratis.proto.RaftProtos.RaftPeerRole;
 import org.apache.ratis.protocol.RaftPeerId;
+import org.apache.ratis.util.JavaUtils;
 import org.apache.ratis.util.Preconditions;
 import org.apache.ratis.util.Timestamp;
 import org.slf4j.Logger;
@@ -87,7 +88,7 @@
     final LeaderStateImpl leader = leaderState.getAndSet(null);
     if (leader == null) {
       if (!allowNull) {
-        throw new NullPointerException("leaderState == null");
+        return JavaUtils.completeExceptionally(new NullPointerException("leaderState == null"));
       }
       return CompletableFuture.completedFuture(null);
     } else {