Put shards failure under a cap flag (elastic#131371)

* Put shards failure under a cap flag

Author: smalyshev

test/external-modules/error-query/src/javaRestTest/java/org/elasticsearch/test/esql/EsqlPartialResultsIT.java

@@ -214,6 +214,10 @@ public void testFailureFromRemote() throws Exception {
     }
 
     public void testAllShardsFailed() throws Exception {
+        assumeTrue(
+            "fail functionality is not enabled",
+            clusterHasCapability("POST", "/_query", List.of(), List.of("fail_if_all_shards_fail")).orElse(false)
+        );
         setupRemoteClusters();
         populateIndices();
         try {
@@ -232,6 +236,26 @@ public void testAllShardsFailed() throws Exception {
         }
     }
 
+    public void testAllShardsFailedOldBehavior() throws Exception {
+        // TODO: drop this once we no longer support the old behavior
+        assumeFalse(
+            "fail functionality is enabled",
+            clusterHasCapability("POST", "/_query", List.of(), List.of("fail_if_all_shards_fail")).orElse(false)
+        );
+        setupRemoteClusters();
+        populateIndices();
+        try {
+            Request request = new Request("POST", "/_query");
+            request.setJsonEntity("{\"query\": \"FROM " + "*:failing*" + " | LIMIT 100\"}");
+            request.addParameter("allow_partial_results", "true");
+            Response resp = client().performRequest(request);
+            Map<String, Object> results = entityAsMap(resp);
+            assertThat(results.get("is_partial"), equalTo(true));
+        } finally {
+            removeRemoteCluster();
+        }
+    }
+
     private void setupRemoteClusters() throws IOException {
         String settings = String.format(Locale.ROOT, """
             {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1263,6 +1263,11 @@ public enum Cap {
          */
         NO_BRACKETS_IN_UNQUOTED_INDEX_NAMES,
 
+        /**
+         * Fail if all shards fail
+         */
+        FAIL_IF_ALL_SHARDS_FAIL(Build.current().isSnapshot()),
+
         /*
          * Cosine vector similarity function
          */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java

@@ -47,6 +47,7 @@
 import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
 import org.elasticsearch.xpack.esql.action.EsqlExecutionInfo;
 import org.elasticsearch.xpack.esql.action.EsqlQueryAction;
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
@@ -548,6 +549,9 @@ private static void updateExecutionInfoAfterCoordinatorOnlyQuery(EsqlExecutionIn
      * which doesn't consider the failures from the remote clusters when skip_unavailable is true.
      */
     static void failIfAllShardsFailed(EsqlExecutionInfo execInfo, List<Page> finalResults) {
+        if (EsqlCapabilities.Cap.FAIL_IF_ALL_SHARDS_FAIL.isEnabled() == false) {
+            return;
+        }
         // do not fail if any final result has results
         if (finalResults.stream().anyMatch(p -> p.getPositionCount() > 0)) {
             return;