Ignore remote ES|QL execution failures when skip_unavailable=true

docs/changelog/116365.yaml

@@ -0,0 +1,5 @@
+pr: 116365
+summary: Ignore remote ES|QL execution failures when skip_unavailable=true
+area: ES|QL
+type: enhancement
+issues: []

test/framework/src/main/java/org/elasticsearch/test/FailingFieldPlugin.java

@@ -24,12 +24,14 @@
 
 public class FailingFieldPlugin extends Plugin implements ScriptPlugin {
 
+    public static final String FAILING_FIELD_LANG = "failing_field";
+
     @Override
     public ScriptEngine getScriptEngine(Settings settings, Collection<ScriptContext<?>> contexts) {
         return new ScriptEngine() {
             @Override
             public String getType() {
-                return "failing_field";
+                return FAILING_FIELD_LANG;
             }
 
             @Override

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/Clusters.java

@@ -31,6 +31,10 @@ public static ElasticsearchCluster remoteCluster() {
     }
 
     public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteCluster) {
+        return localCluster(remoteCluster, true);
+    }
+
+    public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteCluster, Boolean skipUnavailable) {
         return ElasticsearchCluster.local()
             .name(LOCAL_CLUSTER_NAME)
             .distribution(DistributionType.DEFAULT)
@@ -40,6 +44,7 @@ public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteClust
             .setting("xpack.license.self_generated.type", "trial")
             .setting("node.roles", "[data,ingest,master,remote_cluster_client]")
             .setting("cluster.remote.remote_cluster.seeds", () -> "\"" + remoteCluster.getTransportEndpoint(0) + "\"")
+            .setting("cluster.remote.remote_cluster.skip_unavailable", skipUnavailable.toString())
             .setting("cluster.remote.connections_per_cluster", "1")
             .shared(true)
             .setting("cluster.routing.rebalance.enable", "none")

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/EsqlRestValidationIT.java

@@ -30,7 +30,7 @@
 @ThreadLeakFilters(filters = TestClustersThreadFilter.class)
 public class EsqlRestValidationIT extends EsqlRestValidationTestCase {
     static ElasticsearchCluster remoteCluster = Clusters.remoteCluster();
-    static ElasticsearchCluster localCluster = Clusters.localCluster(remoteCluster);
+    static ElasticsearchCluster localCluster = Clusters.localCluster(remoteCluster, false);
 
     @ClassRule
     public static TestRule clusterRule = RuleChain.outerRule(remoteCluster).around(localCluster);

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/EsqlRestValidationSkipUnavailableIT.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.ccq;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
+
+import org.elasticsearch.test.TestClustersThreadFilter;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.junit.ClassRule;
+import org.junit.rules.RuleChain;
+import org.junit.rules.TestRule;
+
+import java.io.IOException;
+
+@ThreadLeakFilters(filters = TestClustersThreadFilter.class)
+public class EsqlRestValidationSkipUnavailableIT extends EsqlRestValidationIT {
+    static ElasticsearchCluster localCluster = Clusters.localCluster(remoteCluster, true);
+
+    @ClassRule
+    public static TestRule clusterRule = RuleChain.outerRule(remoteCluster).around(localCluster);
+
+    @Override
+    protected String getTestRestCluster() {
+        return localCluster.getHttpAddresses();
+    }
+
+    @Override
+    protected void assertErrorMessageMaybe(String indexName, String errorMessage, int statusCode) throws IOException {
+        assertValidRequestOnIndices(new String[] { indexName });
+    }
+
+}

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/RequestIndexFilteringIT.java

@@ -30,7 +30,7 @@
 public class RequestIndexFilteringIT extends RequestIndexFilteringTestCase {
 
     static ElasticsearchCluster remoteCluster = Clusters.remoteCluster();
-    static ElasticsearchCluster localCluster = Clusters.localCluster(remoteCluster);
+    static ElasticsearchCluster localCluster = Clusters.localCluster(remoteCluster, false);
 
     @ClassRule
     public static TestRule clusterRule = RuleChain.outerRule(remoteCluster).around(localCluster);

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/EsqlRestValidationTestCase.java

@@ -87,26 +87,30 @@ public void testInexistentIndexNameWithoutWildcard() throws IOException {
 
     public void testExistentIndexWithoutWildcard() throws IOException {
         for (String indexName : existentIndexWithoutWildcard) {
-            assertErrorMessage(indexName, "\"reason\" : \"no such index [inexistent]\"", 404);
+            assertErrorMessageMaybe(indexName, "\"reason\" : \"no such index [inexistent]\"", 404);
         }
     }
 
     public void testExistentIndexWithWildcard() throws IOException {
         assertValidRequestOnIndices(existentIndexWithWildcard);
     }
 
+    protected void assertErrorMessageMaybe(String indexName, String errorMessage, int statusCode) throws IOException {
+        assertErrorMessage(indexName, errorMessage, statusCode);
+    }
+
     public void testAlias() throws IOException {
         createAlias();
 
         for (String indexName : existentAliasWithoutWildcard) {
-            assertErrorMessage(indexName, "\"reason\" : \"no such index [inexistent]\"", 404);
+            assertErrorMessageMaybe(indexName, "\"reason\" : \"no such index [inexistent]\"", 404);
         }
         assertValidRequestOnIndices(existentAliasWithWildcard);
 
         deleteAlias();
     }
 
-    private void assertErrorMessages(String[] indices, String errorMessage, int statusCode) throws IOException {
+    protected void assertErrorMessages(String[] indices, String errorMessage, int statusCode) throws IOException {
         for (String indexName : indices) {
             assertErrorMessage(indexName, errorMessage + "[" + clusterSpecificIndexName(indexName) + "]", statusCode);
         }
@@ -116,7 +120,7 @@ protected String clusterSpecificIndexName(String indexName) {
         return indexName;
     }
 
-    private void assertErrorMessage(String indexName, String errorMessage, int statusCode) throws IOException {
+    protected void assertErrorMessage(String indexName, String errorMessage, int statusCode) throws IOException {
         var specificName = clusterSpecificIndexName(indexName);
         final var request = createRequest(specificName);
         ResponseException exc = expectThrows(ResponseException.class, () -> client().performRequest(request));
@@ -138,7 +142,7 @@ private Request createRequest(String indexName) throws IOException {
         return request;
     }
 
-    private void assertValidRequestOnIndices(String[] indices) throws IOException {
+    protected void assertValidRequestOnIndices(String[] indices) throws IOException {
         for (String indexName : indices) {
             final var request = createRequest(clusterSpecificIndexName(indexName));
             Response response = client().performRequest(request);

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractPausableIntegTestCase.java

@@ -22,6 +22,8 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+import static org.elasticsearch.xpack.esql.action.AbstractPauseFieldPlugin.PAUSE_FIELD_LANG;
+
 /** A pausable testcase. Subclasses extend this testcase to simulate slow running queries.
  *
  * Uses the evaluation of a runtime field in the mappings "pause_me" of type long, along
@@ -64,7 +66,7 @@ public void setupIndex() throws IOException {
             mapping.startObject("pause_me");
             {
                 mapping.field("type", "long");
-                mapping.startObject("script").field("source", "").field("lang", "pause").endObject();
+                mapping.startObject("script").field("source", "").field("lang", PAUSE_FIELD_LANG).endObject();
             }
             mapping.endObject();
         }

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractPauseFieldPlugin.java

@@ -28,6 +28,8 @@
  */
 public abstract class AbstractPauseFieldPlugin extends Plugin implements ScriptPlugin {
 
+    public static final String PAUSE_FIELD_LANG = "pause";
+
     // Called when the engine enters the execute() method.
     protected void onStartExecute() {}
 
@@ -39,7 +41,7 @@ public ScriptEngine getScriptEngine(Settings settings, Collection<ScriptContext<
         return new ScriptEngine() {
             @Override
             public String getType() {
-                return "pause";
+                return PAUSE_FIELD_LANG;
             }
 
             @Override

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterAsyncQueryIT.java

@@ -412,7 +412,7 @@ void populateRemoteIndicesWithRuntimeMapping(String clusterAlias) throws IOExcep
             mapping.startObject("const");
             {
                 mapping.field("type", "long");
-                mapping.startObject("script").field("source", "").field("lang", "pause").endObject();
+                mapping.startObject("script").field("source", "").field("lang", AbstractPauseFieldPlugin.PAUSE_FIELD_LANG).endObject();
             }
             mapping.endObject();
         }

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClustersCancellationIT.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.action;
 
+import org.elasticsearch.Build;
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.cancel.TransportCancelTasksAction;
@@ -20,6 +21,7 @@
 import org.elasticsearch.compute.operator.exchange.ExchangeService;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.test.AbstractMultiClustersTestCase;
 import org.elasticsearch.transport.TransportService;
@@ -31,6 +33,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.getValuesList;
@@ -39,10 +42,16 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.instanceOf;
 
 public class CrossClustersCancellationIT extends AbstractMultiClustersTestCase {
     private static final String REMOTE_CLUSTER = "cluster-a";
 
+    @Override
+    protected boolean reuseClusters() {
+        return false;
+    }
+
     @Override
     protected List<String> remoteClusterAlias() {
         return List.of(REMOTE_CLUSTER);
@@ -57,6 +66,11 @@ protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
         return plugins;
     }
 
+    @Override
+    protected Map<String, Boolean> skipUnavailableForRemoteClusters() {
+        return Map.of("cluster-a", true, "cluster-b", false);
+    }
+
     public static class InternalExchangePlugin extends Plugin {
         @Override
         public List<Setting<?>> getSettings() {
@@ -82,7 +96,7 @@ private void createRemoteIndex(int numDocs) throws Exception {
             mapping.startObject("const");
             {
                 mapping.field("type", "long");
-                mapping.startObject("script").field("source", "").field("lang", "pause").endObject();
+                mapping.startObject("script").field("source", "").field("lang", AbstractPauseFieldPlugin.PAUSE_FIELD_LANG).endObject();
             }
             mapping.endObject();
         }
@@ -96,6 +110,26 @@ private void createRemoteIndex(int numDocs) throws Exception {
         bulk.get();
     }
 
+    private void createLocalIndex(int numDocs) throws Exception {
+        XContentBuilder mapping = JsonXContent.contentBuilder().startObject();
+        mapping.startObject("runtime");
+        {
+            mapping.startObject("const");
+            {
+                mapping.field("type", "long");
+            }
+            mapping.endObject();
+        }
+        mapping.endObject();
+        mapping.endObject();
+        client(LOCAL_CLUSTER).admin().indices().prepareCreate("test").setMapping(mapping).get();
+        BulkRequestBuilder bulk = client(LOCAL_CLUSTER).prepareBulk("test").setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        for (int i = 0; i < numDocs; i++) {
+            bulk.add(new IndexRequest().source("const", i));
+        }
+        bulk.get();
+    }
+
     public void testCancel() throws Exception {
         createRemoteIndex(between(10, 100));
         EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
@@ -208,4 +242,92 @@ public void testTasks() throws Exception {
         }
         requestFuture.actionGet(30, TimeUnit.SECONDS).close();
     }
+
+    // Check that cancelling remote task with skip_unavailable=true produces partial
+    public void testCancelSkipUnavailable() throws Exception {
+        createRemoteIndex(between(10, 100));
+        EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
+        request.query("FROM *:test | STATS total=sum(const) | LIMIT 1");
+        request.pragmas(randomPragmas());
+        request.includeCCSMetadata(true);
+        PlainActionFuture<EsqlQueryResponse> requestFuture = new PlainActionFuture<>();
+        client().execute(EsqlQueryAction.INSTANCE, request, requestFuture);
+        assertTrue(SimplePauseFieldPlugin.startEmitting.await(30, TimeUnit.SECONDS));
+        List<TaskInfo> rootTasks = new ArrayList<>();
+        assertBusy(() -> {
+            List<TaskInfo> tasks = client(REMOTE_CLUSTER).admin()
+                .cluster()
+                .prepareListTasks()
+                .setActions(ComputeService.CLUSTER_ACTION_NAME)
+                .get()
+                .getTasks();
+            assertThat(tasks, hasSize(1));
+            rootTasks.addAll(tasks);
+        });
+        var cancelRequest = new CancelTasksRequest().setTargetTaskId(rootTasks.get(0).taskId()).setReason("remote failed");
+        client(REMOTE_CLUSTER).execute(TransportCancelTasksAction.TYPE, cancelRequest);
+        try {
+            assertBusy(() -> {
+                List<TaskInfo> drivers = client(REMOTE_CLUSTER).admin()
+                    .cluster()
+                    .prepareListTasks()
+                    .setActions(DriverTaskRunner.ACTION_NAME)
+                    .get()
+                    .getTasks();
+                assertThat(drivers.size(), greaterThanOrEqualTo(1));
+                for (TaskInfo driver : drivers) {
+                    assertTrue(driver.cancelled());
+                }
+            });
+        } finally {
+            SimplePauseFieldPlugin.allowEmitting.countDown();
+        }
+        var resp = requestFuture.actionGet();
+        EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+
+        assertNotNull(executionInfo);
+        EsqlExecutionInfo.Cluster cluster = executionInfo.getCluster(REMOTE_CLUSTER);
+
+        assertThat(cluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.PARTIAL));
+        assertThat(cluster.getFailures().size(), equalTo(1));
+        assertThat(cluster.getFailures().get(0).getCause(), instanceOf(TaskCancelledException.class));
+    }
+
+    // Check that closing remote node with skip_unavailable=true produces partial
+    public void testCloseSkipUnavailable() throws Exception {
+        assumeTrue("Only snapshot builds have delay()", Build.current().isSnapshot());
+        createRemoteIndex(between(1000, 5000));
+        createLocalIndex(100);
+        EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
+        request.query("""
+                FROM test*,cluster-a:test* METADATA _index
+                | EVAL cluster=MV_FIRST(SPLIT(_index, ":"))
+                | WHERE CASE(cluster == "cluster-a", delay(1ms), true)
+                | STATS total = sum(const) | LIMIT 1
+            """);
+        request.pragmas(randomPragmas());
+        var requestFuture = client().execute(EsqlQueryAction.INSTANCE, request);
+        assertTrue(SimplePauseFieldPlugin.startEmitting.await(30, TimeUnit.SECONDS));
+        SimplePauseFieldPlugin.allowEmitting.countDown();
+        cluster(REMOTE_CLUSTER).close();
+        try (var resp = requestFuture.actionGet()) {
+            EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+            assertNotNull(executionInfo);
+
+            List<List<Object>> values = getValuesList(resp);
+            assertThat(values.get(0).size(), equalTo(1));
+            // We can't be sure of the exact value here as we don't know if any data from remote came in, but all local data should be there
+            assertThat((long) values.get(0).get(0), greaterThanOrEqualTo(4950L));
+
+            EsqlExecutionInfo.Cluster cluster = executionInfo.getCluster(REMOTE_CLUSTER);
+            EsqlExecutionInfo.Cluster localCluster = executionInfo.getCluster(LOCAL_CLUSTER);
+
+            assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL));
+            assertThat(localCluster.getSuccessfulShards(), equalTo(1));
+
+            assertThat(cluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.PARTIAL));
+            assertThat(cluster.getSuccessfulShards(), equalTo(0));
+            assertThat(cluster.getFailures().size(), equalTo(1));
+        }
+    }
 }