RATIS-1874. Add notifyLeaderReady function in IStateMachine (apache#906)

ratis-server-api/src/main/java/org/apache/ratis/statemachine/StateMachine.java

@@ -228,6 +228,11 @@ default void notifyFollowerSlowness(RoleInfoProto leaderInfo) {}
      * Notify {@link StateMachine} that this server is no longer the leader.
      */
     default void notifyNotLeader(Collection<TransactionContext> pendingEntries) throws IOException {}
+
+    /**
+     * Notify the {@link StateMachine} that this server becomes ready after changed to leader.
+     */
+    default void notifyLeaderReady() {}
   }
 
   /**

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

@@ -315,6 +315,7 @@ private List<FollowerInfo> getFollowerInfos(PeerConfiguration peers) {
 
   private final int stagingCatchupGap;
   private final long placeHolderIndex;
+  private final AtomicBoolean isReady = new AtomicBoolean();
   private final RaftServerMetricsImpl raftServerMetrics;
   private final LogAppenderMetrics logAppenderMetrics;
   private final long followerMaxGapThreshold;
@@ -382,7 +383,14 @@ LogEntryProto start() {
   }
 
   boolean isReady() {
-    return server.getState().getLastAppliedIndex() >= placeHolderIndex;
+    return isReady.get();
+  }
+
+  void checkReady(LogEntryProto entry) {
+    if (entry.getTerm() == currentTerm && entry.getIndex() == placeHolderIndex) {
+      isReady.set(true);
+      server.getStateMachine().leaderEvent().notifyLeaderReady();
+    }
   }
 
   void stop() {

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

@@ -1802,6 +1802,7 @@ CompletableFuture<Message> applyLogToStateMachine(LogEntryProto next) throws Raf
       // the new conf in the metadata file and notify the StateMachine.
       state.writeRaftConfiguration(next);
       stateMachine.event().notifyConfigurationChanged(next.getTerm(), next.getIndex(), next.getConfigurationEntry());
+      role.getLeaderState().ifPresent(leader -> leader.checkReady(next));
     } else if (next.hasStateMachineLogEntry()) {
       // check whether there is a TransactionContext because we are the leader.
       TransactionContext trx = role.getLeaderState()