RATIS-2050. Add creationGap param to snapshot management API

ratis-client/src/main/java/org/apache/ratis/client/api/SnapshotManagementApi.java

@@ -27,6 +27,24 @@
  */
 public interface SnapshotManagementApi {
 
-  /** trigger create snapshot file. */
-  RaftClientReply create(long timeoutMs) throws IOException;
+  /** The same as create(0, timeoutMs). */
+  default RaftClientReply create(long timeoutMs) throws IOException {
+    return create(0, timeoutMs);
+  }
+
+    /** The same as create(force? 1 : 0, timeoutMs). */
+  default RaftClientReply create(boolean force, long timeoutMs) throws IOException {
+    return create(force? 1 : 0, timeoutMs);
+  }
+
+    /**
+   * Trigger to create a snapshot.
+   *
+   * @param creationGap When (creationGap > 0) and (astAppliedIndex - lastSnapshotIndex < creationGap),
+   *                    return lastSnapshotIndex; otherwise, take a new snapshot and then return its index.
+   *                    When creationGap == 0, use the server configured value as the creationGap.
+   * @return a reply.  When {@link RaftClientReply#isSuccess()} is true,
+   *         {@link RaftClientReply#getLogIndex()} is the snapshot index fulfilling the operation.
+   */
+  RaftClientReply create(long creationGap, long timeoutMs) throws IOException;
 }

ratis-client/src/main/java/org/apache/ratis/client/impl/ClientProtoUtils.java

@@ -659,7 +659,8 @@ static SnapshotManagementRequest toSnapshotManagementRequest(SnapshotManagementR
     switch(p.getOpCase()) {
       case CREATE:
         return SnapshotManagementRequest.newCreate(clientId, serverId,
-            ProtoUtils.toRaftGroupId(m.getRaftGroupId()), m.getCallId(), m.getTimeoutMs());
+            ProtoUtils.toRaftGroupId(m.getRaftGroupId()), m.getCallId(), m.getTimeoutMs(),
+            p.getCreate().getCreationGap());
       default:
         throw new IllegalArgumentException("Unexpected op " + p.getOpCase() + " in " + p);
     }
@@ -671,7 +672,7 @@ static SnapshotManagementRequestProto toSnapshotManagementRequestProto(
         .setRpcRequest(toRaftRpcRequestProtoBuilder(request));
     final SnapshotManagementRequest.Create create = request.getCreate();
     if (create != null) {
-      b.setCreate(SnapshotCreateRequestProto.newBuilder().build());
+      b.setCreate(SnapshotCreateRequestProto.newBuilder().setCreationGap(create.getCreationGap()).build());
     }
     return b.build();
   }

ratis-client/src/main/java/org/apache/ratis/client/impl/SnapshotManagementImpl.java

@@ -37,9 +37,10 @@ class SnapshotManagementImpl implements SnapshotManagementApi {
   }
 
   @Override
-  public RaftClientReply create(long timeoutMs) throws IOException {
+  public RaftClientReply create(long creationGap, long timeoutMs) throws IOException {
     final long callId = CallId.getAndIncrement();
     return client.io().sendRequestWithRetry(() -> SnapshotManagementRequest.newCreate(client.getId(),
-        Optional.ofNullable(server).orElseGet(client::getLeaderId), client.getGroupId(), callId, timeoutMs));
+        Optional.ofNullable(server).orElseGet(client::getLeaderId),
+        client.getGroupId(), callId, timeoutMs, creationGap));
   }
 }

ratis-common/src/main/java/org/apache/ratis/protocol/SnapshotManagementRequest.java

@@ -24,7 +24,16 @@ public final class SnapshotManagementRequest extends RaftClientRequest {
   public abstract static class Op {
 
   }
-  public static class Create extends Op {
+
+  public static final class Create extends Op {
+    private final long creationGap;
+    private Create(long creationGap) {
+      this.creationGap = creationGap;
+    }
+
+    public long getCreationGap() {
+      return creationGap;
+    }
 
     @Override
     public String toString() {
@@ -35,8 +44,13 @@ public String toString() {
 
   public static SnapshotManagementRequest newCreate(ClientId clientId,
       RaftPeerId serverId, RaftGroupId groupId, long callId, long timeoutMs) {
+    return newCreate(clientId, serverId, groupId, callId, timeoutMs, 0);
+  }
+
+  public static SnapshotManagementRequest newCreate(ClientId clientId,
+      RaftPeerId serverId, RaftGroupId groupId, long callId, long timeoutMs, long creationGap) {
     return new SnapshotManagementRequest(clientId,
-        serverId, groupId, callId, timeoutMs,new SnapshotManagementRequest.Create());
+        serverId, groupId, callId, timeoutMs, new SnapshotManagementRequest.Create(creationGap));
   }
 
   private final Op op;

ratis-proto/src/main/proto/Raft.proto

@@ -470,7 +470,7 @@ message SnapshotManagementRequestProto {
 }
 
 message SnapshotCreateRequestProto {
-
+  uint64 creationGap = 1;
 }
 
 message StartLeaderElectionRequestProto {

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

@@ -1223,9 +1223,10 @@ CompletableFuture<RaftClientReply> takeSnapshotAsync(SnapshotManagementRequest r
     LOG.info("{}: takeSnapshotAsync {}", getMemberId(), request);
     assertLifeCycleState(LifeCycle.States.RUNNING);
     assertGroup(getMemberId(), request);
+    Preconditions.assertNotNull(request.getCreate(), "create");
 
-    //TODO(liuyaolong): get the gap value from shell command
-    long minGapValue = RaftServerConfigKeys.Snapshot.creationGap(proxy.getProperties());
+    final long creationGap = request.getCreate().getCreationGap();
+    long minGapValue = creationGap > 0? creationGap : RaftServerConfigKeys.Snapshot.creationGap(proxy.getProperties());
     final long lastSnapshotIndex = Optional.ofNullable(stateMachine.getLatestSnapshot())
         .map(SnapshotInfo::getIndex)
         .orElse(0L);