Fix OpenPIT with older versions

Author: luigidellaquila

x-pack/plugin/eql/qa/mixed-node/src/javaRestTest/java/org/elasticsearch/xpack/eql/qa/mixed_node/EqlSearchIT.java

@@ -77,19 +77,19 @@ public void cleanUpIndex() throws IOException {
     }
 
     public void testEventsWithRequestToOldNodes() throws Exception {
-        assertEventsQueryOnNodes(bwcNodes);
+        assertEventsQueryOnNodes(bwcNodes, true);
     }
 
     public void testEventsWithRequestToUpgradedNodes() throws Exception {
-        assertEventsQueryOnNodes(newNodes);
+        assertEventsQueryOnNodes(newNodes, false);
     }
 
     public void testSequencesWithRequestToOldNodes() throws Exception {
-        assertSequncesQueryOnNodes(bwcNodes);
+        assertSequncesQueryOnNodes(bwcNodes, true);
     }
 
     public void testSequencesWithRequestToUpgradedNodes() throws Exception {
-        assertSequncesQueryOnNodes(newNodes);
+        assertSequncesQueryOnNodes(newNodes, false);
     }
 
     /**
@@ -268,7 +268,7 @@ public void testMultiValueFields() throws Exception {
         assertTrue(testedFunctions.containsAll(availableFunctions));
     }
 
-    private void assertEventsQueryOnNodes(List<TestNode> nodesList) throws Exception {
+    private void assertEventsQueryOnNodes(List<TestNode> nodesList, boolean oldNodes) throws Exception {
         final String event = randomEvent();
         Map<String, Object> expectedResponse = prepareEventsTestData(event);
         try (
@@ -278,12 +278,25 @@ private void assertEventsQueryOnNodes(List<TestNode> nodesList) throws Exception
             String filterPath = "filter_path=hits.events._source.@timestamp,hits.events._source.event_type,hits.events._source.sequence";
 
             Request request = new Request("POST", index + "/_eql/search?" + filterPath);
-            request.setJsonEntity("{\"query\":\"" + event + " where true\",\"size\":15}");
-            assertBusy(() -> { assertResponse(expectedResponse, runEql(client, request)); });
+            StringBuilder payload = new StringBuilder("{\"query\":\"" + event + " where true\",\"size\":15");
+            // Old versions don't support this option
+            if (oldNodes == false) {
+                if (randomBoolean()) {
+                    payload.append(", \"allow_partial_search_results\": " + randomBoolean());
+                }
+                if (randomBoolean()) {
+                    payload.append(", \"allow_partial_sequence_results\": " + randomBoolean());
+                }
+            }
+            payload.append("}");
+            request.setJsonEntity(payload.toString());
+            assertBusy(() -> {
+                assertResponse(expectedResponse, runEql(client, request));
+            });
         }
     }
 
-    private void assertSequncesQueryOnNodes(List<TestNode> nodesList) throws Exception {
+    private void assertSequncesQueryOnNodes(List<TestNode> nodesList, boolean oldNodes) throws Exception {
         Map<String, Object> expectedResponse = prepareSequencesTestData();
         try (
             RestClient client = buildClient(restClientSettings(), nodesList.stream().map(TestNode::publishAddress).toArray(HttpHost[]::new))
@@ -294,7 +307,19 @@ private void assertSequncesQueryOnNodes(List<TestNode> nodesList) throws Excepti
             String filter = "{\"range\":{\"@timestamp\":{\"gte\":\"1970-05-01\"}}}";
 
             Request request = new Request("POST", index + "/_eql/search?" + filterPath);
-            request.setJsonEntity("{\"query\":\"" + query + "\",\"filter\":" + filter + "}");
+
+            StringBuilder payload = new StringBuilder("{\"query\":\"" + query + "\",\"filter\":" + filter);
+            // Old versions don't support this option
+            if (oldNodes == false) {
+                if (randomBoolean()) {
+                    payload.append(", \"allow_partial_search_results\": " + randomBoolean());
+                }
+                if (randomBoolean()) {
+                    payload.append(", \"allow_partial_sequence_results\": " + randomBoolean());
+                }
+            }
+            payload.append("}");
+            request.setJsonEntity(payload.toString());
             assertBusy(() -> { assertResponse(expectedResponse, runEql(client, request)); });
         }
     }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/PITAwareQueryClient.java

@@ -7,12 +7,14 @@
 
 package org.elasticsearch.xpack.eql.execution.search;
 
+import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.ClosePointInTimeRequest;
 import org.elasticsearch.action.search.ClosePointInTimeResponse;
 import org.elasticsearch.action.search.MultiSearchRequest;
 import org.elasticsearch.action.search.MultiSearchResponse;
 import org.elasticsearch.action.search.OpenPointInTimeRequest;
+import org.elasticsearch.action.search.OpenPointInTimeResponse;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.TransportClosePointInTimeAction;
@@ -140,10 +142,33 @@ private <Response> void openPIT(ActionListener<Response> listener, Runnable runn
             .keepAlive(keepAlive)
             .allowPartialSearchResults(allowPartialSearchResults);
         request.indexFilter(filter);
-        client.execute(TransportOpenPointInTimeAction.TYPE, request, listener.delegateFailureAndWrap((l, r) -> {
-            pitId = r.getPointInTimeId();
-            runnable.run();
-        }));
+
+        client.execute(TransportOpenPointInTimeAction.TYPE, request, new ActionListener<>() {
+            @Override
+            public void onResponse(OpenPointInTimeResponse r) {
+                try {
+                    pitId = r.getPointInTimeId();
+                    runnable.run();
+                } catch (Exception e) {
+                    onFailure(e);
+                }
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                if (allowPartialSearchResults
+                    && e instanceof ElasticsearchStatusException
+                    && e.getMessage()
+                        .contains("The [allow_partial_search_results] parameter cannot be used while the cluster is still upgrading.")) {
+                    // This is for pre-8.16
+                    // We cannot use allow_partial_search_results during upgrades, so let's try without and hope to get no shard failures,
+                    // it's the best we can do, the query would fail anyway
+                    openPIT(listener, runnable, false);
+                } else {
+                    listener.onFailure(e);
+                }
+            }
+        });
     }
 
     @Override