Validate PIT on _msearch (elastic#63167) (elastic#74460)

This change ensures that we validate point in times provided by individual search
requests in _msearch.

Relates elastic#63132

Author: dnhatn

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/350_point_in_time.yml

@@ -171,8 +171,8 @@ setup:
 ---
 "msearch":
   - skip:
-      version: " - 7.99.99"
-      reason: "After backport: 7.9.99 => point in time is introduced in 7.10"
+      version: " - 7.9.99"
+      reason: "point in time is introduced in 7.10"
   - do:
       open_point_in_time:
         index: "t*"

server/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.settings.Settings;
@@ -80,7 +81,7 @@ public String getName() {
 
     @Override
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
-        MultiSearchRequest multiSearchRequest = parseRequest(request, allowExplicitIndex);
+        MultiSearchRequest multiSearchRequest = parseRequest(request,  client.getNamedWriteableRegistry(), allowExplicitIndex);
 
         // Emit a single deprecation message if any search request contains types.
         for (SearchRequest searchRequest : multiSearchRequest.requests()) {
@@ -98,7 +99,9 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
     /**
      * Parses a {@link RestRequest} body and returns a {@link MultiSearchRequest}
      */
-    public static MultiSearchRequest parseRequest(RestRequest restRequest, boolean allowExplicitIndex) throws IOException {
+    public static MultiSearchRequest parseRequest(RestRequest restRequest,
+                                                  NamedWriteableRegistry namedWriteableRegistry,
+                                                  boolean allowExplicitIndex) throws IOException {
         MultiSearchRequest multiRequest = new MultiSearchRequest();
         IndicesOptions indicesOptions = IndicesOptions.fromRequest(restRequest, multiRequest.indicesOptions());
         multiRequest.indicesOptions(indicesOptions);
@@ -123,6 +126,13 @@ public static MultiSearchRequest parseRequest(RestRequest restRequest, boolean a
         parseMultiLineRequest(restRequest, multiRequest.indicesOptions(), allowExplicitIndex, (searchRequest, parser) -> {
             searchRequest.source(SearchSourceBuilder.fromXContent(parser, false));
             RestSearchAction.checkRestTotalHits(restRequest, searchRequest);
+            if (searchRequest.pointInTimeBuilder() != null) {
+                RestSearchAction.preparePointInTime(searchRequest, restRequest, namedWriteableRegistry);
+            } else {
+                searchRequest.setCcsMinimizeRoundtrips(
+                    restRequest.paramAsBoolean("ccs_minimize_roundtrips", searchRequest.isCcsMinimizeRoundtrips())
+                );
+            }
             multiRequest.add(searchRequest);
         });
         List<SearchRequest> requests = multiRequest.requests();

server/src/test/java/org/elasticsearch/action/search/MultiSearchRequestTests.java

@@ -95,7 +95,7 @@ public void testFailWithUnknownKey() {
         FakeRestRequest restRequest = new FakeRestRequest.Builder(xContentRegistry())
             .withContent(new BytesArray(requestContent), XContentType.JSON).build();
         IllegalArgumentException ex = expectThrows(IllegalArgumentException.class,
-            () -> RestMultiSearchAction.parseRequest(restRequest, true));
+            () -> RestMultiSearchAction.parseRequest(restRequest, null, true));
         assertEquals("key [unknown_key] is not supported in the metadata section", ex.getMessage());
     }
 
@@ -104,7 +104,7 @@ public void testSimpleAddWithCarriageReturn() throws Exception {
             "{\"query\" : {\"match_all\" :{}}}\r\n";
         FakeRestRequest restRequest = new FakeRestRequest.Builder(xContentRegistry())
             .withContent(new BytesArray(requestContent), XContentType.JSON).build();
-        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, true);
+        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, null, true);
         assertThat(request.requests().size(), equalTo(1));
         assertThat(request.requests().get(0).indices()[0], equalTo("test"));
         assertThat(request.requests().get(0).indicesOptions(),
@@ -119,7 +119,7 @@ public void testDefaultIndicesOptions() throws IOException {
             .withContent(new BytesArray(requestContent), XContentType.JSON)
             .withParams(Collections.singletonMap("ignore_unavailable", "true"))
             .build();
-        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, true);
+        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, null, true);
         assertThat(request.requests().size(), equalTo(1));
         assertThat(request.requests().get(0).indices()[0], equalTo("test"));
         assertThat(request.requests().get(0).indicesOptions(),
@@ -254,13 +254,13 @@ public void testMsearchTerminatedByNewline() throws Exception {
         RestRequest restRequest = new FakeRestRequest.Builder(xContentRegistry())
                 .withContent(new BytesArray(mserchAction.getBytes(StandardCharsets.UTF_8)), XContentType.JSON).build();
         IllegalArgumentException expectThrows = expectThrows(IllegalArgumentException.class,
-                () -> RestMultiSearchAction.parseRequest(restRequest, true));
+                () -> RestMultiSearchAction.parseRequest(restRequest, null, true));
         assertEquals("The msearch request must be terminated by a newline [\n]", expectThrows.getMessage());
 
         String mserchActionWithNewLine = mserchAction + "\n";
         RestRequest restRequestWithNewLine = new FakeRestRequest.Builder(xContentRegistry())
                 .withContent(new BytesArray(mserchActionWithNewLine.getBytes(StandardCharsets.UTF_8)), XContentType.JSON).build();
-        MultiSearchRequest msearchRequest = RestMultiSearchAction.parseRequest(restRequestWithNewLine, true);
+        MultiSearchRequest msearchRequest = RestMultiSearchAction.parseRequest(restRequestWithNewLine, null, true);
         assertEquals(3, msearchRequest.requests().size());
     }
 

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/data_stream/10_data_stream_resolvability.yml

@@ -664,7 +664,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -675,7 +674,7 @@
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3 }
+  - match: {hits.total.value: 3 }
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: "/\\.ds-simple-data-stream1-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000001/" }
   - match: {hits.hits.0._id: "123" }
@@ -684,7 +683,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -695,7 +693,7 @@
           pit:
             id: "$point_in_time_id"
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: "/\\.ds-simple-data-stream1-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000001/" }
   - match: {hits.hits.0._id: "5" }
@@ -704,7 +702,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -716,7 +713,7 @@
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: "/\\.ds-simple-data-stream1-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000001/" }
   - match: {hits.hits.0._id: "1" }
@@ -725,7 +722,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -737,7 +733,7 @@
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 0 }
 
   - do: