feat(pit): Elasticsearch _pit support with engine-aware gating, plus integration and request-building test coverage

src/main/java/org/codelibs/fesen/client/action/HttpCloseIndexAction.java

@@ -29,6 +29,7 @@
 import org.opensearch.action.admin.indices.close.CloseIndexResponse.IndexResult;
 import org.opensearch.action.admin.indices.close.CloseIndexResponse.ShardResult;
 import org.opensearch.action.admin.indices.close.CloseIndexResponse.ShardResult.Failure;
+import org.opensearch.action.support.ActiveShardCount;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.index.Index;
 import org.opensearch.core.xcontent.XContentParser;
@@ -85,6 +86,10 @@ protected CurlRequest getCurlRequest(final CloseIndexRequest request) {
         if (request.masterNodeTimeout() != null) {
             curlRequest.param("master_timeout", request.masterNodeTimeout().toString());
         }
+        if (!ActiveShardCount.DEFAULT.equals(request.waitForActiveShards())) {
+            curlRequest.param("wait_for_active_shards", getActiveShardsCountString(request.waitForActiveShards()));
+        }
+        appendIndicesOptions(curlRequest, request.indicesOptions());
         return curlRequest;
     }
 

src/main/java/org/codelibs/fesen/client/action/HttpCreatePitAction.java

@@ -16,12 +16,15 @@
 package org.codelibs.fesen.client.action;
 
 import java.io.IOException;
+import java.util.Map;
 
 import org.codelibs.curl.CurlRequest;
+import org.codelibs.fesen.client.EngineInfo.EngineType;
 import org.codelibs.fesen.client.HttpClient;
 import org.opensearch.action.search.CreatePitAction;
 import org.opensearch.action.search.CreatePitRequest;
 import org.opensearch.action.search.CreatePitResponse;
+import org.opensearch.action.search.ShardSearchFailure;
 import org.opensearch.action.support.IndicesOptions;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.xcontent.XContentParser;
@@ -30,10 +33,18 @@
  * Handles the Create Point-in-Time (PIT) API over HTTP, opening a PIT reader
  * context over one or more indices that can be reused across search requests.
  *
- * <p>This action targets the OpenSearch REST API only. The
- * {@code /{index}/_search/point_in_time} endpoint is OpenSearch-specific;
- * Elasticsearch uses a different {@code _pit} endpoint and Elasticsearch 7 has
- * no PIT support at all. No Elasticsearch adaptation is provided.
+ * <p>PIT is supported on OpenSearch 2.x and later via the
+ * {@code POST /{index}/_search/point_in_time} endpoint, and on Elasticsearch
+ * 7.10+ and 8.x via the {@code POST /{index}/_pit} endpoint. The two backends
+ * differ in both endpoint and query parameters (OpenSearch accepts
+ * {@code allow_partial_pit_creation}, which the Elasticsearch {@code _pit}
+ * endpoint rejects), and their response bodies differ, so the engine reported by
+ * {@link HttpClient#getEngineInfo()} selects the request and response handling.
+ *
+ * <p>PIT is not available on OpenSearch 1.x, and the engine cannot be
+ * determined for {@link EngineType#UNKNOWN}; in both cases
+ * {@link #execute(CreatePitRequest, ActionListener)} fails the listener with an
+ * {@link UnsupportedOperationException}.
  */
 public class HttpCreatePitAction extends HttpAction {
 
@@ -54,13 +65,23 @@ public HttpCreatePitAction(final HttpClient client, final CreatePitAction action
     /**
      * Executes the create PIT request and notifies the listener with the response.
      *
+     * <p>Fails the listener with an {@link UnsupportedOperationException} when the
+     * backend is OpenSearch 1.x or an unknown engine.
+     *
      * @param request the create PIT request containing the target indices and keep-alive
      * @param listener the listener notified with the response or a failure
      */
     public void execute(final CreatePitRequest request, final ActionListener<CreatePitResponse> listener) {
+        final EngineType type = client.getEngineInfo().getType();
+        if (type == EngineType.OPENSEARCH1 || type == EngineType.UNKNOWN) {
+            listener.onFailure(new UnsupportedOperationException(
+                    "Point-In-Time is not supported on " + type + " over HTTP (requires OpenSearch 2.x+ or Elasticsearch 7.10+)"));
+            return;
+        }
+        final boolean elasticsearch = type == EngineType.ELASTICSEARCH7 || type == EngineType.ELASTICSEARCH8;
         getCurlRequest(request).execute(response -> {
             try (final XContentParser parser = createParser(response)) {
-                final CreatePitResponse pitResponse = fromXContent(parser);
+                final CreatePitResponse pitResponse = elasticsearch ? parseEsCreate(parser) : fromXContent(parser);
                 listener.onResponse(pitResponse);
             } catch (final Exception e) {
                 listener.onFailure(toOpenSearchException(response, e));
@@ -69,7 +90,7 @@ public void execute(final CreatePitRequest request, final ActionListener<CreateP
     }
 
     /**
-     * Parses a create PIT response from the given XContent parser.
+     * Parses an OpenSearch create PIT response from the given XContent parser.
      *
      * @param parser the parser positioned at the response body
      * @return the parsed create PIT response
@@ -80,13 +101,76 @@ protected CreatePitResponse fromXContent(final XContentParser parser) throws IOE
     }
 
     /**
-     * Builds the HTTP request for the create PIT API endpoint.
+     * Parses an Elasticsearch open PIT response ({@code {"id":...,"_shards":{...}}}) into a
+     * {@link CreatePitResponse}. Elasticsearch does not report a creation time, so the current
+     * system time is used, and no shard failures are reconstructed.
+     *
+     * @param parser the parser positioned at the response body
+     * @return the parsed create PIT response
+     * @throws IOException if parsing the response fails
+     */
+    CreatePitResponse parseEsCreate(final XContentParser parser) throws IOException {
+        if (parser.currentToken() == null) {
+            parser.nextToken();
+        }
+        final Map<String, Object> map = parser.map();
+        final String id = (String) map.get("id");
+        if (id == null) {
+            // Not a successful open-PIT body (e.g. an error response routed through the success
+            // path); surface it as a failure instead of returning a response with a null id.
+            throw new IOException("Unexpected create PIT response (missing \"id\"): " + map);
+        }
+        int total = 0;
+        int successful = 0;
+        int skipped = 0;
+        int failed = 0;
+        final Object shardsObj = map.get("_shards");
+        if (shardsObj instanceof Map) {
+            @SuppressWarnings("unchecked")
+            final Map<String, Object> shards = (Map<String, Object>) shardsObj;
+            total = toInt(shards.get("total"));
+            successful = toInt(shards.get("successful"));
+            skipped = toInt(shards.get("skipped"));
+            failed = toInt(shards.get("failed"));
+        }
+        return new CreatePitResponse(id, System.currentTimeMillis(), total, successful, skipped, failed, ShardSearchFailure.EMPTY_ARRAY);
+    }
+
+    private static int toInt(final Object value) {
+        return value instanceof Number ? ((Number) value).intValue() : 0;
+    }
+
+    /**
+     * Builds the HTTP request for the create PIT API endpoint. The endpoint and query
+     * parameters differ between OpenSearch and Elasticsearch, selected by the engine
+     * reported by {@link HttpClient#getEngineInfo()}.
      *
      * @param request the create PIT request
      * @return the configured curl request
      */
     protected CurlRequest getCurlRequest(final CreatePitRequest request) {
-        // RestCreatePitAction: POST /{index}/_search/point_in_time
+        final EngineType type = client.getEngineInfo().getType();
+        if (type == EngineType.ELASTICSEARCH7 || type == EngineType.ELASTICSEARCH8) {
+            // Elasticsearch Open PIT API: POST /{index}/_pit
+            // Note: the Elasticsearch _pit endpoint does not accept allow_partial_pit_creation /
+            // allow_partial_search_results (unrecognized parameter -> HTTP 400), so it is not sent.
+            final CurlRequest curlRequest = client.getCurlRequest(POST, "/_pit", request.indices());
+            if (request.getKeepAlive() != null) {
+                curlRequest.param("keep_alive", request.getKeepAlive().getStringRep());
+            }
+            if (request.getPreference() != null) {
+                curlRequest.param("preference", request.getPreference());
+            }
+            if (request.getRouting() != null) {
+                curlRequest.param("routing", request.getRouting());
+            }
+            final IndicesOptions esIndicesOptions = request.indicesOptions();
+            if (esIndicesOptions != null) {
+                appendIndicesOptions(curlRequest, esIndicesOptions);
+            }
+            return curlRequest;
+        }
+        // OpenSearch RestCreatePitAction: POST /{index}/_search/point_in_time
         final CurlRequest curlRequest = client.getCurlRequest(POST, "/_search/point_in_time", request.indices());
         if (request.getKeepAlive() != null) {
             curlRequest.param("keep_alive", request.getKeepAlive().getStringRep());

src/main/java/org/codelibs/fesen/client/action/HttpDeleteIndexAction.java

@@ -74,6 +74,7 @@ protected CurlRequest getCurlRequest(final DeleteIndexRequest request) {
         if (request.masterNodeTimeout() != null) {
             curlRequest.param("master_timeout", request.masterNodeTimeout().toString());
         }
+        appendIndicesOptions(curlRequest, request.indicesOptions());
         return curlRequest;
     }
 }

src/main/java/org/codelibs/fesen/client/action/HttpDeletePitAction.java

@@ -16,12 +16,16 @@
 package org.codelibs.fesen.client.action;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import org.codelibs.curl.CurlRequest;
+import org.codelibs.fesen.client.EngineInfo.EngineType;
 import org.codelibs.fesen.client.HttpClient;
 import org.opensearch.OpenSearchException;
 import org.opensearch.action.search.DeletePitAction;
+import org.opensearch.action.search.DeletePitInfo;
 import org.opensearch.action.search.DeletePitRequest;
 import org.opensearch.action.search.DeletePitResponse;
 import org.opensearch.common.xcontent.json.JsonXContent;
@@ -35,15 +39,20 @@
  * Handles the Delete Point-in-Time (PIT) API over HTTP, releasing one or more
  * PIT reader contexts.
  *
- * <p>When the request targets the special {@code _all} identifier the delete-all
- * endpoint {@code DELETE /_search/point_in_time/_all} is used with no body;
- * otherwise the PIT identifiers are sent as a {@code {"pit_id":[...]}} body to
- * {@code DELETE /_search/point_in_time}.
+ * <p>On OpenSearch 2.x+ a delete-by-id request sends a {@code {"pit_id":[...]}}
+ * body to {@code DELETE /_search/point_in_time}, and the special {@code _all}
+ * identifier targets {@code DELETE /_search/point_in_time/_all} with no body. On
+ * Elasticsearch 7.10+ and 8.x a delete-by-id request sends a {@code {"id":"..."}}
+ * body to {@code DELETE /_pit} (the endpoint accepts a single identifier only, so
+ * a request carrying multiple identifiers fails the listener with an
+ * {@link UnsupportedOperationException}); Elasticsearch has no delete-all
+ * endpoint, so a delete-all request likewise fails the listener with an
+ * {@link UnsupportedOperationException}.
  *
- * <p>This action targets the OpenSearch REST API only. The
- * {@code _search/point_in_time} endpoint is OpenSearch-specific; Elasticsearch
- * uses a different {@code _pit} endpoint and Elasticsearch 7 has no PIT support
- * at all. No Elasticsearch adaptation is provided.
+ * <p>PIT is not available on OpenSearch 1.x, and the engine cannot be determined
+ * for {@link EngineType#UNKNOWN}; in both cases
+ * {@link #execute(DeletePitRequest, ActionListener)} fails the listener with an
+ * {@link UnsupportedOperationException}.
  */
 public class HttpDeletePitAction extends HttpAction {
 
@@ -67,17 +76,39 @@ public HttpDeletePitAction(final HttpClient client, final DeletePitAction action
     /**
      * Executes the delete PIT request and notifies the listener with the response.
      *
+     * <p>Fails the listener with an {@link UnsupportedOperationException} when the
+     * backend is OpenSearch 1.x or an unknown engine, or when a delete-all request
+     * targets Elasticsearch (which has no delete-all endpoint).
+     *
      * @param request the delete PIT request containing the PIT identifiers to release
      * @param listener the listener notified with the response or a failure
      */
     public void execute(final DeletePitRequest request, final ActionListener<DeletePitResponse> listener) {
+        final EngineType type = client.getEngineInfo().getType();
+        if (type == EngineType.OPENSEARCH1 || type == EngineType.UNKNOWN) {
+            listener.onFailure(new UnsupportedOperationException(
+                    "Point-In-Time is not supported on " + type + " over HTTP (requires OpenSearch 2.x+ or Elasticsearch 7.10+)"));
+            return;
+        }
+        final boolean elasticsearch = type == EngineType.ELASTICSEARCH7 || type == EngineType.ELASTICSEARCH8;
+        if (elasticsearch && isDeleteAll(request)) {
+            listener.onFailure(new UnsupportedOperationException("Delete-all PITs is not supported on Elasticsearch over HTTP"));
+            return;
+        }
+        if (elasticsearch && request.getPitIds() != null && request.getPitIds().size() > 1) {
+            // The Elasticsearch _pit endpoint accepts a single id per request (an array body is
+            // rejected with HTTP 400); callers must delete multiple PITs individually.
+            listener.onFailure(new UnsupportedOperationException(
+                    "Deleting multiple PITs in a single request is not supported on Elasticsearch over HTTP; delete them individually"));
+            return;
+        }
         final CurlRequest curlRequest = getCurlRequest(request);
         if (!isDeleteAll(request)) {
-            curlRequest.body(buildBody(request));
+            curlRequest.body(elasticsearch ? buildEsBody(request) : buildBody(request));
         }
         curlRequest.execute(response -> {
             try (final XContentParser parser = createParser(response)) {
-                final DeletePitResponse pitResponse = fromXContent(parser);
+                final DeletePitResponse pitResponse = elasticsearch ? parseEsDelete(parser, request) : fromXContent(parser);
                 listener.onResponse(pitResponse);
             } catch (final Exception e) {
                 listener.onFailure(toOpenSearchException(response, e));
@@ -86,7 +117,7 @@ public void execute(final DeletePitRequest request, final ActionListener<DeleteP
     }
 
     /**
-     * Parses a delete PIT response from the given XContent parser.
+     * Parses an OpenSearch delete PIT response from the given XContent parser.
      *
      * @param parser the parser positioned at the response body
      * @return the parsed delete PIT response
@@ -97,14 +128,49 @@ protected DeletePitResponse fromXContent(final XContentParser parser) throws IOE
     }
 
     /**
-     * Builds the HTTP request for the delete PIT API endpoint. A delete-all request
-     * targets the {@code /_all} endpoint; otherwise the plain endpoint is used.
+     * Parses an Elasticsearch close PIT response ({@code {"succeeded":...,"num_freed":...}}) into
+     * a {@link DeletePitResponse}. Elasticsearch returns only an aggregate {@code succeeded} flag
+     * and no per-identifier list, so the flag is applied to each requested identifier.
+     *
+     * @param parser the parser positioned at the response body
+     * @param request the originating delete PIT request providing the identifiers
+     * @return the parsed delete PIT response
+     * @throws IOException if parsing the response fails
+     */
+    DeletePitResponse parseEsDelete(final XContentParser parser, final DeletePitRequest request) throws IOException {
+        if (parser.currentToken() == null) {
+            parser.nextToken();
+        }
+        final Map<String, Object> map = parser.map();
+        if (!map.containsKey("succeeded")) {
+            // Not a successful close-PIT body (e.g. an error response routed through the success
+            // path); surface it as a failure instead of reporting every id as failed.
+            throw new IOException("Unexpected delete PIT response (missing \"succeeded\"): " + map);
+        }
+        final boolean succeeded = Boolean.TRUE.equals(map.get("succeeded"));
+        final List<DeletePitInfo> results = new ArrayList<>();
+        for (final String pitId : request.getPitIds()) {
+            results.add(new DeletePitInfo(succeeded, pitId));
+        }
+        return new DeletePitResponse(results);
+    }
+
+    /**
+     * Builds the HTTP request for the delete PIT API endpoint. The endpoint is selected by the
+     * engine reported by {@link HttpClient#getEngineInfo()}: Elasticsearch uses {@code /_pit},
+     * OpenSearch uses {@code /_search/point_in_time} (or its {@code /_all} variant for a
+     * delete-all request).
      *
      * @param request the delete PIT request
      * @return the configured curl request
      */
     protected CurlRequest getCurlRequest(final DeletePitRequest request) {
-        // RestDeletePitAction
+        final EngineType type = client.getEngineInfo().getType();
+        if (type == EngineType.ELASTICSEARCH7 || type == EngineType.ELASTICSEARCH8) {
+            // Elasticsearch Close PIT API: DELETE /_pit
+            return client.getCurlRequest(DELETE, "/_pit");
+        }
+        // OpenSearch RestDeletePitAction
         if (isDeleteAll(request)) {
             return client.getCurlRequest(DELETE, "/_search/point_in_time/_all");
         }
@@ -124,7 +190,8 @@ protected static boolean isDeleteAll(final DeletePitRequest request) {
     }
 
     /**
-     * Serializes the delete PIT request body as a {@code {"pit_id":[...]}} JSON object.
+     * Serializes the delete PIT request body as a {@code {"pit_id":[...]}} JSON object
+     * for OpenSearch.
      *
      * @param request the delete PIT request
      * @return the JSON request body
@@ -138,4 +205,24 @@ protected String buildBody(final DeletePitRequest request) {
             throw new OpenSearchException("Failed to build a request.", e);
         }
     }
+
+    /**
+     * Serializes the delete PIT request body as a {@code {"id":"..."}} JSON object for
+     * Elasticsearch, which accepts a single identifier only. Callers with multiple identifiers
+     * are rejected earlier in {@link #execute(DeletePitRequest, ActionListener)}.
+     *
+     * @param request the delete PIT request
+     * @return the JSON request body
+     */
+    protected String buildEsBody(final DeletePitRequest request) {
+        try (final XContentBuilder builder = JsonXContent.contentBuilder()) {
+            builder.startObject();
+            builder.field("id", request.getPitIds().get(0));
+            builder.endObject();
+            builder.flush();
+            return BytesReference.bytes(builder).utf8ToString();
+        } catch (final IOException e) {
+            throw new OpenSearchException("Failed to build a request.", e);
+        }
+    }
 }

src/main/java/org/codelibs/fesen/client/action/HttpFlushAction.java

@@ -70,6 +70,7 @@ protected CurlRequest getCurlRequest(final FlushRequest request) {
         final CurlRequest curlRequest = client.getCurlRequest(POST, "/_flush", request.indices());
         curlRequest.param("wait_if_ongoing", Boolean.toString(request.waitIfOngoing()));
         curlRequest.param("force", Boolean.toString(request.force()));
+        appendIndicesOptions(curlRequest, request.indicesOptions());
         return curlRequest;
     }
 }