Skip to content

Commit d3c8ef5

Browse files
smalyshevsmalyshev
committed
Add tests
1 parent dcd0d78 commit d3c8ef5

File tree

6 files changed

+203
-2
lines changed

6 files changed

+203
-2
lines changed

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,14 @@
2424

2525
public class FailingFieldPlugin extends Plugin implements ScriptPlugin {
2626

27+
public static final String FAILING_FIELD_LANG = "failing_field";
28+
2729
@Override
2830
public ScriptEngine getScriptEngine(Settings settings, Collection<ScriptContext<?>> contexts) {
2931
return new ScriptEngine() {
3032
@Override
3133
public String getType() {
32-
return "failing_field";
34+
return FAILING_FIELD_LANG;
3335
}
3436

3537
@Override

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@ public static ElasticsearchCluster remoteCluster() {
3131
}
3232

3333
public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteCluster) {
34+
return localCluster(remoteCluster, true);
35+
}
36+
37+
public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteCluster, Boolean skipUnavailable) {
3438
return ElasticsearchCluster.local()
3539
.name(LOCAL_CLUSTER_NAME)
3640
.distribution(DistributionType.DEFAULT)
@@ -41,6 +45,7 @@ public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteClust
4145
.setting("node.roles", "[data,ingest,master,remote_cluster_client]")
4246
.setting("cluster.remote.remote_cluster.seeds", () -> "\"" + remoteCluster.getTransportEndpoint(0) + "\"")
4347
.setting("cluster.remote.connections_per_cluster", "1")
48+
.setting("cluster.remote." + REMOTE_CLUSTER_NAME + ".skip_unavailable", skipUnavailable.toString())
4449
.shared(true)
4550
.setting("cluster.routing.rebalance.enable", "none")
4651
.build();

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111

1212
import org.apache.http.HttpHost;
1313
import org.elasticsearch.Version;
14+
import org.elasticsearch.client.Request;
15+
import org.elasticsearch.client.Response;
1416
import org.elasticsearch.client.RestClient;
1517
import org.elasticsearch.core.IOUtils;
1618
import org.elasticsearch.test.TestClustersThreadFilter;
@@ -50,6 +52,20 @@ public static void closeRemoteClients() throws IOException {
5052
}
5153
}
5254

55+
@Before
56+
public void randomizeSkipUnavailable() throws IOException {
57+
Request request = new Request("PUT", "_cluster/settings");
58+
request.setJsonEntity("""
59+
{
60+
"persistent": {
61+
"cluster.remote.remote_cluster.skip_unavailable": $SKIP_UNAVAILABLE$
62+
}
63+
}
64+
""".replace("$SKIP_UNAVAILABLE$", String.valueOf(randomBoolean())));
65+
Response response = client().performRequest(request);
66+
assertOK(response);
67+
}
68+
5369
@Override
5470
protected String clusterSpecificIndexName(String pattern) {
5571
StringJoiner sj = new StringJoiner(",");

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

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77

88
package org.elasticsearch.xpack.esql.action;
99

10+
import org.elasticsearch.Build;
1011
import org.elasticsearch.action.ActionFuture;
1112
import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
1213
import org.elasticsearch.action.admin.cluster.node.tasks.cancel.TransportCancelTasksAction;
@@ -20,6 +21,7 @@
2021
import org.elasticsearch.compute.operator.exchange.ExchangeService;
2122
import org.elasticsearch.core.TimeValue;
2223
import org.elasticsearch.plugins.Plugin;
24+
import org.elasticsearch.tasks.TaskCancelledException;
2325
import org.elasticsearch.tasks.TaskInfo;
2426
import org.elasticsearch.test.AbstractMultiClustersTestCase;
2527
import org.elasticsearch.transport.TransportService;
@@ -39,6 +41,7 @@
3941
import static org.hamcrest.Matchers.equalTo;
4042
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
4143
import static org.hamcrest.Matchers.hasSize;
44+
import static org.hamcrest.Matchers.instanceOf;
4245

4346
public class CrossClustersCancellationIT extends AbstractMultiClustersTestCase {
4447
private static final String REMOTE_CLUSTER = "cluster-a";
@@ -75,6 +78,11 @@ public void resetPlugin() {
7578
SimplePauseFieldPlugin.resetPlugin();
7679
}
7780

81+
@Override
82+
protected boolean reuseClusters() {
83+
return false;
84+
}
85+
7886
private void createRemoteIndex(int numDocs) throws Exception {
7987
XContentBuilder mapping = JsonXContent.contentBuilder().startObject();
8088
mapping.startObject("runtime");
@@ -96,6 +104,26 @@ private void createRemoteIndex(int numDocs) throws Exception {
96104
bulk.get();
97105
}
98106

107+
private void createLocalIndex(int numDocs) throws Exception {
108+
XContentBuilder mapping = JsonXContent.contentBuilder().startObject();
109+
mapping.startObject("runtime");
110+
{
111+
mapping.startObject("const");
112+
{
113+
mapping.field("type", "long");
114+
}
115+
mapping.endObject();
116+
}
117+
mapping.endObject();
118+
mapping.endObject();
119+
client(LOCAL_CLUSTER).admin().indices().prepareCreate("test").setMapping(mapping).get();
120+
BulkRequestBuilder bulk = client(LOCAL_CLUSTER).prepareBulk("test").setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
121+
for (int i = 0; i < numDocs; i++) {
122+
bulk.add(new IndexRequest().source("const", i));
123+
}
124+
bulk.get();
125+
}
126+
99127
public void testCancel() throws Exception {
100128
createRemoteIndex(between(10, 100));
101129
EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
@@ -208,4 +236,93 @@ public void testTasks() throws Exception {
208236
}
209237
requestFuture.actionGet(30, TimeUnit.SECONDS).close();
210238
}
239+
240+
// Check that cancelling remote task with skip_unavailable=true produces partial
241+
public void testCancelSkipUnavailable() throws Exception {
242+
createRemoteIndex(between(10, 100));
243+
EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
244+
request.query("FROM *:test | STATS total=sum(const) | LIMIT 1");
245+
request.pragmas(randomPragmas());
246+
request.includeCCSMetadata(true);
247+
PlainActionFuture<EsqlQueryResponse> requestFuture = new PlainActionFuture<>();
248+
client().execute(EsqlQueryAction.INSTANCE, request, requestFuture);
249+
assertTrue(SimplePauseFieldPlugin.startEmitting.await(30, TimeUnit.SECONDS));
250+
List<TaskInfo> rootTasks = new ArrayList<>();
251+
assertBusy(() -> {
252+
List<TaskInfo> tasks = client(REMOTE_CLUSTER).admin()
253+
.cluster()
254+
.prepareListTasks()
255+
.setActions(ComputeService.CLUSTER_ACTION_NAME)
256+
.get()
257+
.getTasks();
258+
assertThat(tasks, hasSize(1));
259+
rootTasks.addAll(tasks);
260+
});
261+
var cancelRequest = new CancelTasksRequest().setTargetTaskId(rootTasks.get(0).taskId()).setReason("remote failed");
262+
client(REMOTE_CLUSTER).execute(TransportCancelTasksAction.TYPE, cancelRequest);
263+
try {
264+
assertBusy(() -> {
265+
List<TaskInfo> drivers = client(REMOTE_CLUSTER).admin()
266+
.cluster()
267+
.prepareListTasks()
268+
.setActions(DriverTaskRunner.ACTION_NAME)
269+
.get()
270+
.getTasks();
271+
assertThat(drivers.size(), greaterThanOrEqualTo(1));
272+
for (TaskInfo driver : drivers) {
273+
assertTrue(driver.cancelled());
274+
}
275+
});
276+
} finally {
277+
SimplePauseFieldPlugin.allowEmitting.countDown();
278+
}
279+
var resp = requestFuture.actionGet();
280+
EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
281+
282+
assertNotNull(executionInfo);
283+
EsqlExecutionInfo.Cluster cluster = executionInfo.getCluster(REMOTE_CLUSTER);
284+
285+
assertThat(cluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.PARTIAL));
286+
assertThat(cluster.getFailures().size(), equalTo(1));
287+
assertThat(cluster.getFailures().get(0).getCause(), instanceOf(TaskCancelledException.class));
288+
}
289+
290+
// Check that closing remote node with skip_unavailable=true produces partial
291+
public void testCloseSkipUnavailable() throws Exception {
292+
// We are using delay() here because closing cluster while inside pause fields doesn't seem to produce clean closure
293+
assumeTrue("Only snapshot builds have delay()", Build.current().isSnapshot());
294+
createRemoteIndex(between(1000, 5000));
295+
createLocalIndex(10);
296+
EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
297+
request.query("""
298+
FROM test*,cluster-a:test* METADATA _index
299+
| EVAL cluster=MV_FIRST(SPLIT(_index, ":"))
300+
| WHERE CASE(cluster == "cluster-a", delay(1ms), true)
301+
| STATS total = sum(const) | LIMIT 1
302+
""");
303+
request.pragmas(randomPragmas());
304+
var requestFuture = client().execute(EsqlQueryAction.INSTANCE, request);
305+
assertTrue(SimplePauseFieldPlugin.startEmitting.await(30, TimeUnit.SECONDS));
306+
SimplePauseFieldPlugin.allowEmitting.countDown();
307+
cluster(REMOTE_CLUSTER).close();
308+
try (var resp = requestFuture.actionGet()) {
309+
EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
310+
assertNotNull(executionInfo);
311+
312+
List<List<Object>> values = getValuesList(resp);
313+
assertThat(values.get(0).size(), equalTo(1));
314+
// 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
315+
assertThat((long) values.get(0).get(0), greaterThanOrEqualTo(45L));
316+
317+
EsqlExecutionInfo.Cluster cluster = executionInfo.getCluster(REMOTE_CLUSTER);
318+
EsqlExecutionInfo.Cluster localCluster = executionInfo.getCluster(LOCAL_CLUSTER);
319+
320+
assertThat(localCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL));
321+
assertThat(localCluster.getSuccessfulShards(), equalTo(1));
322+
323+
assertThat(cluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.PARTIAL));
324+
assertThat(cluster.getSuccessfulShards(), equalTo(0));
325+
assertThat(cluster.getFailures().size(), equalTo(1));
326+
}
327+
}
211328
}

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

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,10 +27,13 @@
2727
import org.elasticsearch.index.query.TermsQueryBuilder;
2828
import org.elasticsearch.plugins.Plugin;
2929
import org.elasticsearch.test.AbstractMultiClustersTestCase;
30+
import org.elasticsearch.test.FailingFieldPlugin;
3031
import org.elasticsearch.test.InternalTestCluster;
3132
import org.elasticsearch.test.XContentTestUtils;
3233
import org.elasticsearch.transport.RemoteClusterAware;
3334
import org.elasticsearch.transport.TransportService;
35+
import org.elasticsearch.xcontent.XContentBuilder;
36+
import org.elasticsearch.xcontent.json.JsonXContent;
3437
import org.elasticsearch.xpack.esql.VerificationException;
3538
import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
3639

@@ -81,6 +84,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
8184
List<Class<? extends Plugin>> plugins = new ArrayList<>(super.nodePlugins(clusterAlias));
8285
plugins.add(EsqlPluginWithEnterpriseOrTrialLicense.class);
8386
plugins.add(InternalExchangePlugin.class);
87+
plugins.add(FailingFieldPlugin.class);
8488
return plugins;
8589
}
8690

@@ -836,6 +840,17 @@ public void testWarnings() throws Exception {
836840
assertTrue(latch.await(30, TimeUnit.SECONDS));
837841
}
838842

843+
// Non-disconnect remote failures still fail the request
844+
public void testRemoteFailureSkipUnavailable() throws IOException {
845+
Map<String, Object> testClusterInfo = setupFailClusters();
846+
String localIndex = (String) testClusterInfo.get("local.index");
847+
String remote1Index = (String) testClusterInfo.get("remote.index");
848+
int localNumShards = (Integer) testClusterInfo.get("local.num_shards");
849+
String q = Strings.format("FROM %s,cluster-a:%s*", localIndex, remote1Index);
850+
IllegalStateException e = expectThrows(IllegalStateException.class, () -> runQuery(q, false));
851+
assertThat(e.getMessage(), containsString("Accessing failing field"));
852+
}
853+
839854
private static void assertClusterMetadataInResponse(EsqlQueryResponse resp, boolean responseExpectMeta) {
840855
try {
841856
final Map<String, Object> esqlResponseAsMap = XContentTestUtils.convertToMap(resp);
@@ -1058,4 +1073,46 @@ private void clearSkipUnavailable() {
10581073
.setPersistentSettings(settingsBuilder.build())
10591074
.get();
10601075
}
1076+
1077+
Map<String, Object> setupFailClusters() throws IOException {
1078+
int numShardsLocal = randomIntBetween(1, 3);
1079+
populateLocalIndices(LOCAL_INDEX, numShardsLocal);
1080+
1081+
int numShardsRemote = randomIntBetween(1, 3);
1082+
populateRemoteIndicesFail(REMOTE_CLUSTER_1, REMOTE_INDEX, numShardsRemote);
1083+
1084+
Map<String, Object> clusterInfo = new HashMap<>();
1085+
clusterInfo.put("local.num_shards", numShardsLocal);
1086+
clusterInfo.put("local.index", LOCAL_INDEX);
1087+
clusterInfo.put("remote.num_shards", numShardsRemote);
1088+
clusterInfo.put("remote.index", REMOTE_INDEX);
1089+
setSkipUnavailable(REMOTE_CLUSTER_1, true);
1090+
return clusterInfo;
1091+
}
1092+
1093+
void populateRemoteIndicesFail(String clusterAlias, String indexName, int numShards) throws IOException {
1094+
Client remoteClient = client(clusterAlias);
1095+
XContentBuilder mapping = JsonXContent.contentBuilder().startObject();
1096+
mapping.startObject("runtime");
1097+
{
1098+
mapping.startObject("fail_me");
1099+
{
1100+
mapping.field("type", "long");
1101+
mapping.startObject("script").field("source", "").field("lang", FailingFieldPlugin.FAILING_FIELD_LANG).endObject();
1102+
}
1103+
mapping.endObject();
1104+
}
1105+
mapping.endObject();
1106+
assertAcked(
1107+
remoteClient.admin()
1108+
.indices()
1109+
.prepareCreate(indexName)
1110+
.setSettings(Settings.builder().put("index.number_of_shards", numShards))
1111+
.setMapping(mapping.endObject())
1112+
);
1113+
1114+
remoteClient.prepareIndex(indexName).setSource("id", 0).get();
1115+
remoteClient.admin().indices().prepareRefresh(indexName).get();
1116+
}
1117+
10611118
}

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@ public void onStartExecute() {
3131

3232
@Override
3333
public boolean onWait() throws InterruptedException {
34-
return allowEmitting.await(30, TimeUnit.SECONDS);
34+
try {
35+
return allowEmitting.await(30, TimeUnit.SECONDS);
36+
} catch (InterruptedException e) {
37+
return true;
38+
}
3539
}
3640
}

0 commit comments

Comments
 (0)