[CELEBORN-1631] Support to apply ratis snapshot operation with RESTfu…

…l api

### What changes were proposed in this pull request?
Sub task of CELEBORN-1628

### Why are the changes needed?

Mapping for ratis-shell command:

```
$ celeborn-ratis sh snapshot create -peers <P0_HOST:P0_PORT,P1_HOST:P1_PORT,P2_HOST:P2_PORT> -peerId <peerId0> [-groupid <RAFT_GROUP_ID>]
```
It is helpful for automation integration.

### Does this PR introduce _any_ user-facing change?

No, it is a new API.

### How was this patch tested?
Integration testing.

<img width="1259" alt="image" src="https://github.com/user-attachments/assets/1f74a899-17f3-41b3-911a-36374bf0fd0b">

Closes #2803 from turboFei/snapshot_create.

Authored-by: Wang, Fei <[email protected]>
Signed-off-by: Shuang <[email protected]>

master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/RatisResource.scala

@@ -25,7 +25,7 @@ import scala.collection.JavaConverters._
 import io.swagger.v3.oas.annotations.media.{Content, Schema}
 import io.swagger.v3.oas.annotations.responses.ApiResponse
 import io.swagger.v3.oas.annotations.tags.Tag
-import org.apache.ratis.protocol.{LeaderElectionManagementRequest, RaftPeerId, TransferLeadershipRequest}
+import org.apache.ratis.protocol.{LeaderElectionManagementRequest, RaftPeerId, SnapshotManagementRequest, TransferLeadershipRequest}
 import org.apache.ratis.rpc.CallId
 
 import org.apache.celeborn.common.internal.Logging
@@ -92,6 +92,31 @@ class RatisResource extends ApiRequestContext with Logging {
     applyElectionOp(new LeaderElectionManagementRequest.Resume)
   }
 
+  @ApiResponse(
+    responseCode = "200",
+    content = Array(new Content(
+      mediaType = MediaType.APPLICATION_JSON,
+      schema = new Schema(implementation = classOf[HandleResponse]))),
+    description = "Trigger the current server to take snapshot.")
+  @POST
+  @Path("/snapshot/create")
+  def createSnapshot(): HandleResponse = ensureMasterHAEnabled(master) {
+    val request = SnapshotManagementRequest.newCreate(
+      ratisServer.getClientId,
+      ratisServer.getServer.getId,
+      ratisServer.getGroupId,
+      CallId.getAndIncrement(),
+      HARaftServer.REQUEST_TIMEOUT_MS)
+    val reply = ratisServer.getServer.snapshotManagement(request)
+    if (reply.isSuccess) {
+      new HandleResponse().success(true).message(
+        s"Successfully create snapshot at ${ratisServer.getLocalAddress}.")
+    } else {
+      new HandleResponse().success(false).message(
+        s"Failed to create snapshot at ${ratisServer.getLocalAddress}. $reply")
+    }
+  }
+
   private def transferLeadership(peerAddress: String): HandleResponse = {
     val newLeaderId = Option(peerAddress).map(getRaftPeerId).orNull
     val op =

openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/master/RatisApi.java

@@ -47,6 +47,73 @@ public RatisApi(ApiClient apiClient) {
     super(apiClient);
   }
 
+  /**
+   * 
+   * Trigger the current server take snapshot.
+   * @return HandleResponse
+   * @throws ApiException if fails to make API call
+   */
+  public HandleResponse createRatisSnapshot() throws ApiException {
+    return this.createRatisSnapshot(Collections.emptyMap());
+  }
+
+
+  /**
+   * 
+   * Trigger the current server take snapshot.
+   * @param additionalHeaders additionalHeaders for this call
+   * @return HandleResponse
+   * @throws ApiException if fails to make API call
+   */
+  public HandleResponse createRatisSnapshot(Map<String, String> additionalHeaders) throws ApiException {
+    Object localVarPostBody = null;
+
+    // create path and map variables
+    String localVarPath = "/api/v1/ratis/snapshot/create";
+
+    StringJoiner localVarQueryStringJoiner = new StringJoiner("&");
+    String localVarQueryParameterBaseName;
+    List<Pair> localVarQueryParams = new ArrayList<Pair>();
+    List<Pair> localVarCollectionQueryParams = new ArrayList<Pair>();
+    Map<String, String> localVarHeaderParams = new HashMap<String, String>();
+    Map<String, String> localVarCookieParams = new HashMap<String, String>();
+    Map<String, Object> localVarFormParams = new HashMap<String, Object>();
+
+
+    localVarHeaderParams.putAll(additionalHeaders);
+
+
+
+    final String[] localVarAccepts = {
+      "application/json"
+    };
+    final String localVarAccept = apiClient.selectHeaderAccept(localVarAccepts);
+
+    final String[] localVarContentTypes = {
+
+    };
+    final String localVarContentType = apiClient.selectHeaderContentType(localVarContentTypes);
+
+    String[] localVarAuthNames = new String[] { "basic" };
+
+    TypeReference<HandleResponse> localVarReturnType = new TypeReference<HandleResponse>() {};
+    return apiClient.invokeAPI(
+        localVarPath,
+        "POST",
+        localVarQueryParams,
+        localVarCollectionQueryParams,
+        localVarQueryStringJoiner.toString(),
+        localVarPostBody,
+        localVarHeaderParams,
+        localVarCookieParams,
+        localVarFormParams,
+        localVarAccept,
+        localVarContentType,
+        localVarAuthNames,
+        localVarReturnType
+    );
+  }
+
   /**
    * 
    * Pause leader election at the current server. Then, the current server would not start a leader election.

openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml

@@ -319,6 +319,20 @@ paths:
               schema:
                 $ref: '#/components/schemas/HandleResponse'
 
+  /api/v1/ratis/snapshot/create:
+    post:
+      tags:
+        - Ratis
+      operationId: createRatisSnapshot
+      description: Trigger the current server take snapshot.
+      responses:
+        "200":
+          description: The request was successful.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HandleResponse'
+
 components:
   schemas:
     ConfigData: