Skip to content

Commit 9bcdb10

Browse files
GalLaloucheGalLalouche
committed
More CR fixes
1 parent 30f5ef1 commit 9bcdb10

File tree

10 files changed

+85
-128
lines changed

10 files changed

+85
-128
lines changed

‎test/framework/src/main/java/org/elasticsearch/test/IntegerMatcher.java‎ renamed to ‎test/framework/src/main/java/org/elasticsearch/test/IntOrLongMatcher.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,16 +20,16 @@
2020
/**
2121
* A type-agnostic way of comparing integer values, not caring if it's a long or an integer.
2222
*/
23-
public abstract sealed class IntegerMatcher<T> extends BaseMatcher<T> {
24-
public static IntegerMatcher<Integer> matches(int expected) {
23+
public abstract sealed class IntOrLongMatcher<T> extends BaseMatcher<T> {
24+
public static IntOrLongMatcher<Integer> matches(int expected) {
2525
return new IntMatcher(expected);
2626
}
2727

28-
public static IntegerMatcher<Long> matches(long expected) {
28+
public static IntOrLongMatcher<Long> matches(long expected) {
2929
return new LongMatcher(expected);
3030
}
3131

32-
private static final class IntMatcher extends IntegerMatcher<Integer> {
32+
private static final class IntMatcher extends IntOrLongMatcher<Integer> {
3333
private final int expected;
3434

3535
private IntMatcher(int expected) {
@@ -51,7 +51,7 @@ public void describeTo(Description description) {
5151
}
5252
}
5353

54-
private static final class LongMatcher extends IntegerMatcher<Long> {
54+
private static final class LongMatcher extends IntOrLongMatcher<Long> {
5555
private final long expected;
5656

5757
LongMatcher(long expected) {

‎x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/async/AsyncTaskManagementService.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,6 +172,8 @@ public AsyncTaskManagementService(
172172
this.threadPool = threadPool;
173173
}
174174

175+
public static String ASYNC_ACTION_SUFFIX = "[a]";
176+
175177
public void asyncExecute(
176178
Request request,
177179
TimeValue waitForCompletionTimeout,
@@ -182,7 +184,7 @@ public void asyncExecute(
182184
String nodeId = clusterService.localNode().getId();
183185
try (var ignored = threadPool.getThreadContext().newTraceContext()) {
184186
@SuppressWarnings("unchecked")
185-
T searchTask = (T) taskManager.register("transport", action + "[a]", new AsyncRequestWrapper(request, nodeId));
187+
T searchTask = (T) taskManager.register("transport", action + ASYNC_ACTION_SUFFIX, new AsyncRequestWrapper(request, nodeId));
186188
boolean operationStarted = false;
187189
try {
188190
operation.execute(

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

Lines changed: 4 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -16,18 +16,15 @@
1616
import org.elasticsearch.client.RequestOptions;
1717
import org.elasticsearch.client.Response;
1818
import org.elasticsearch.client.ResponseException;
19-
import org.elasticsearch.client.ResponseListener;
2019
import org.elasticsearch.client.WarningsHandler;
2120
import org.elasticsearch.common.bytes.BytesArray;
2221
import org.elasticsearch.common.io.Streams;
2322
import org.elasticsearch.common.settings.Settings;
24-
import org.elasticsearch.common.xcontent.XContentHelper;
2523
import org.elasticsearch.core.CheckedConsumer;
2624
import org.elasticsearch.core.Nullable;
2725
import org.elasticsearch.core.TimeValue;
2826
import org.elasticsearch.logging.LogManager;
2927
import org.elasticsearch.logging.Logger;
30-
import org.elasticsearch.tasks.TaskId;
3128
import org.elasticsearch.test.ListMatcher;
3229
import org.elasticsearch.test.rest.ESRestTestCase;
3330
import org.elasticsearch.xcontent.ToXContent;
@@ -42,11 +39,9 @@
4239

4340
import java.io.ByteArrayOutputStream;
4441
import java.io.IOException;
45-
import java.io.InputStream;
4642
import java.io.InputStreamReader;
4743
import java.io.OutputStream;
4844
import java.nio.charset.StandardCharsets;
49-
import java.time.Duration;
5045
import java.time.ZoneId;
5146
import java.util.ArrayList;
5247
import java.util.Arrays;
@@ -73,7 +68,6 @@
7368
import static org.hamcrest.Matchers.emptyOrNullString;
7469
import static org.hamcrest.Matchers.equalTo;
7570
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
76-
import static org.hamcrest.Matchers.hasKey;
7771
import static org.hamcrest.Matchers.is;
7872
import static org.hamcrest.Matchers.not;
7973
import static org.hamcrest.Matchers.nullValue;
@@ -1375,56 +1369,6 @@ public void testAsyncGetWithoutContentType() throws IOException {
13751369

13761370
}
13771371

1378-
public void testListApi_noRunningQueries_returnsAnObject() throws Exception {
1379-
Request request = prepareListQueriesRequest();
1380-
Response response = performRequest(request);
1381-
assertThat(entityToMap(response.getEntity(), XContentType.JSON), is(Map.of("queries", Map.of())));
1382-
}
1383-
1384-
public void testListApi_runningQuery_returnsQueriesObject() throws Exception {
1385-
bulkLoadTestData(1);
1386-
String query = fromIndex() + " | keep keyword, integer | where delay(10s) | limit 100 ";
1387-
var builder = requestObjectBuilder().query(query);
1388-
Request request = prepareRequest(SYNC);
1389-
String mediaType = attachBody(builder.build(), request);
1390-
RequestOptions.Builder options = request.getOptions().toBuilder();
1391-
options.addHeader("Content-Type", mediaType);
1392-
options.addHeader("Accept", mediaType);
1393-
request.setOptions(options);
1394-
client().performRequestAsync(request, new ResponseListener() {
1395-
@Override
1396-
public void onSuccess(Response response) {}
1397-
1398-
@Override
1399-
public void onFailure(Exception exception) {}
1400-
});
1401-
Thread.sleep(Duration.ofSeconds(5));
1402-
Response response = performRequest(prepareListQueriesRequest());
1403-
@SuppressWarnings("unchecked")
1404-
var listResult = (Map<String, Map<String, Object>>) EsqlTestUtils.singleValue(
1405-
entityToMap(response.getEntity(), XContentType.JSON).values()
1406-
);
1407-
var taskId = new TaskId(EsqlTestUtils.singleValue(listResult.keySet()));
1408-
var queryFromListResult = EsqlTestUtils.singleValue(listResult.values());
1409-
assertThat(queryFromListResult.get("id"), is((int) taskId.getId()));
1410-
assertThat(queryFromListResult.get("node"), is(taskId.getNodeId()));
1411-
assertThat(queryFromListResult.get("query"), is(query));
1412-
assertThat(queryFromListResult, hasKey("start_time_millis"));
1413-
assertThat(queryFromListResult, hasKey("running_time_nanos"));
1414-
1415-
response = performRequest(prepareGetQueryRequest(taskId));
1416-
@SuppressWarnings("unchecked")
1417-
Map<String, Object> getQueryResult = entityToMap(response.getEntity(), XContentType.JSON);
1418-
assertThat(getQueryResult.get("id"), is((int) taskId.getId()));
1419-
assertThat(getQueryResult.get("node"), is(taskId.getNodeId()));
1420-
assertThat(getQueryResult.get("query"), is(query));
1421-
assertThat(getQueryResult.get("start_time_millis"), is(queryFromListResult.get("start_time_millis")));
1422-
assertThat(getQueryResult, hasKey("running_time_nanos"));
1423-
assertThat(getQueryResult, hasKey("coordinating_node"));
1424-
assertThat(getQueryResult, hasKey("data_nodes"));
1425-
Thread.sleep(Duration.ofSeconds(5));
1426-
}
1427-
14281372
protected static Request prepareRequestWithOptions(RequestObjectBuilder requestObject, Mode mode) throws IOException {
14291373
requestObject.build();
14301374
Request request = prepareRequest(mode);
@@ -1455,15 +1399,11 @@ static Map<String, Object> removeAsyncProperties(Map<String, Object> map) {
14551399
}
14561400

14571401
protected static Map<String, Object> entityToMap(HttpEntity entity, XContentType expectedContentType) throws IOException {
1458-
try (InputStream content = entity.getContent()) {
1459-
XContentType xContentType = XContentType.fromMediaType(entity.getContentType().getValue());
1460-
assertEquals(expectedContentType, xContentType);
1461-
var map = XContentHelper.convertToMap(xContentType.xContent(), content, false);
1462-
if (shouldLog()) {
1463-
LOGGER.info("entity={}", map);
1464-
}
1465-
return map;
1402+
var result = EsqlTestUtils.entityToMap(entity, expectedContentType);
1403+
if (shouldLog()) {
1404+
LOGGER.info("entity={}", result);
14661405
}
1406+
return result;
14671407
}
14681408

14691409
static void addAsyncParameters(RequestObjectBuilder requestObject, boolean keepOnCompletion) throws IOException {

‎x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java‎

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

88
package org.elasticsearch.xpack.esql;
99

10+
import org.apache.http.HttpEntity;
1011
import org.apache.lucene.document.InetAddressPoint;
1112
import org.apache.lucene.sandbox.document.HalfFloatPoint;
1213
import org.apache.lucene.util.BytesRef;
@@ -20,6 +21,7 @@
2021
import org.elasticsearch.common.regex.Regex;
2122
import org.elasticsearch.common.settings.Settings;
2223
import org.elasticsearch.common.util.BigArrays;
24+
import org.elasticsearch.common.xcontent.XContentHelper;
2325
import org.elasticsearch.compute.data.AggregateMetricDoubleBlockBuilder;
2426
import org.elasticsearch.compute.data.BlockFactory;
2527
import org.elasticsearch.compute.data.BlockUtils;
@@ -39,6 +41,7 @@
3941
import org.elasticsearch.test.ESTestCase;
4042
import org.elasticsearch.transport.RemoteTransportException;
4143
import org.elasticsearch.transport.TransportService;
44+
import org.elasticsearch.xcontent.XContentType;
4245
import org.elasticsearch.xcontent.json.JsonXContent;
4346
import org.elasticsearch.xpack.esql.action.EsqlQueryResponse;
4447
import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
@@ -858,4 +861,16 @@ public static <T> T singleValue(Collection<T> collection) {
858861
assertThat(collection, hasSize(1));
859862
return collection.iterator().next();
860863
}
864+
865+
public static Map<String, Object> jsonEntityToMap(HttpEntity entity) throws IOException {
866+
return entityToMap(entity, XContentType.JSON);
867+
}
868+
869+
public static Map<String, Object> entityToMap(HttpEntity entity, XContentType expectedContentType) throws IOException {
870+
try (InputStream content = entity.getContent()) {
871+
XContentType xContentType = XContentType.fromMediaType(entity.getContentType().getValue());
872+
assertEquals(expectedContentType, xContentType);
873+
return XContentHelper.convertToMap(xContentType.xContent(), content, false /* ordered */);
874+
}
875+
}
861876
}

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import org.elasticsearch.xpack.core.async.DeleteAsyncResultRequest;
2121
import org.elasticsearch.xpack.core.async.GetAsyncResultRequest;
2222
import org.elasticsearch.xpack.core.async.TransportDeleteAsyncResultAction;
23+
import org.elasticsearch.xpack.esql.core.async.AsyncTaskManagementService;
2324
import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
2425
import org.hamcrest.core.IsEqual;
2526

@@ -136,7 +137,11 @@ public void testGetAsyncWhileQueryTaskIsBeingCancelled() throws Exception {
136137
.toList();
137138
assertThat(tasks.size(), greaterThanOrEqualTo(1));
138139
});
139-
client().admin().cluster().prepareCancelTasks().setActions(EsqlQueryAction.NAME + "[a]").get();
140+
client().admin()
141+
.cluster()
142+
.prepareCancelTasks()
143+
.setActions(EsqlQueryAction.NAME + AsyncTaskManagementService.ASYNC_ACTION_SUFFIX)
144+
.get();
140145
assertBusy(() -> {
141146
List<TaskInfo> tasks = getEsqlQueryTasks().stream().filter(TaskInfo::cancelled).toList();
142147
assertThat(tasks, not(empty()));

‎x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlListQuerriesActionIT.java‎ renamed to ‎x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlListQueriesActionIT.java‎

Lines changed: 16 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -7,33 +7,27 @@
77

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

10-
import org.apache.http.HttpEntity;
1110
import org.elasticsearch.client.Request;
12-
import org.elasticsearch.common.xcontent.XContentHelper;
11+
import org.elasticsearch.client.Response;
1312
import org.elasticsearch.tasks.TaskId;
14-
import org.elasticsearch.test.IntegerMatcher;
13+
import org.elasticsearch.test.IntOrLongMatcher;
1514
import org.elasticsearch.test.MapMatcher;
16-
import org.elasticsearch.xcontent.XContentType;
1715
import org.elasticsearch.xpack.core.async.GetAsyncResultRequest;
1816
import org.elasticsearch.xpack.esql.EsqlTestUtils;
1917

20-
import java.io.IOException;
21-
import java.io.InputStream;
2218
import java.util.List;
2319
import java.util.Map;
2420
import java.util.concurrent.TimeUnit;
2521

2622
import static org.elasticsearch.core.TimeValue.timeValueSeconds;
23+
import static org.elasticsearch.xpack.esql.EsqlTestUtils.jsonEntityToMap;
2724
import static org.hamcrest.Matchers.allOf;
2825
import static org.hamcrest.Matchers.everyItem;
2926
import static org.hamcrest.Matchers.is;
3027
import static org.hamcrest.Matchers.isA;
3128

32-
/**
33-
* Individual tests for specific aspects of the async query API.
34-
*/
35-
public class EsqlListQuerriesActionIT extends AbstractPausableIntegTestCase {
36-
public static final String QUERY = "from test | stats sum(pause_me)";
29+
public class EsqlListQueriesActionIT extends AbstractPausableIntegTestCase {
30+
private static final String QUERY = "from test | stats sum(pause_me)";
3731

3832
@Override
3933
protected boolean addMockHttpTransport() {
@@ -43,34 +37,34 @@ protected boolean addMockHttpTransport() {
4337
public void testNoRunningQueries() throws Exception {
4438
var request = new Request("GET", "/_query/queries");
4539
var response = getRestClient().performRequest(request);
46-
assertThat(entityToMap(response.getEntity()), is(Map.of("queries", Map.of())));
40+
assertThat(jsonEntityToMap(response.getEntity()), is(Map.of("queries", Map.of())));
4741
}
4842

4943
public void testRunningQueries() throws Exception {
5044
String id = null;
5145
try (var initialResponse = sendAsyncQuery()) {
5246
id = initialResponse.asyncExecutionId().get();
53-
// FIXME(gal, NOCOMMIT) more copy paste
47+
5448
var getResultsRequest = new GetAsyncResultRequest(id);
5549
getResultsRequest.setWaitForCompletionTimeout(timeValueSeconds(1));
5650
client().execute(EsqlAsyncGetResultAction.INSTANCE, getResultsRequest).get().close();
57-
var request = new Request("GET", "/_query/queries");
58-
var response = getRestClient().performRequest(request);
51+
Response listResponse = getRestClient().performRequest(new Request("GET", "/_query/queries"));
5952
@SuppressWarnings("unchecked")
60-
var listResult = (Map<String, Map<String, Object>>) EsqlTestUtils.singleValue(entityToMap(response.getEntity()).values());
53+
var listResult = (Map<String, Map<String, Object>>) EsqlTestUtils.singleValue(
54+
jsonEntityToMap(listResponse.getEntity()).values()
55+
);
6156
var taskId = new TaskId(EsqlTestUtils.singleValue(listResult.keySet()));
6257
MapMatcher basicMatcher = MapMatcher.matchesMap()
6358
.entry("node", is(taskId.getNodeId()))
64-
.entry("id", IntegerMatcher.matches(taskId.getId()))
59+
.entry("id", IntOrLongMatcher.matches(taskId.getId()))
6560
.entry("query", is(QUERY))
66-
.entry("start_time_millis", IntegerMatcher.isIntOrLong())
67-
.entry("running_time_nanos", IntegerMatcher.isIntOrLong());
61+
.entry("start_time_millis", IntOrLongMatcher.isIntOrLong())
62+
.entry("running_time_nanos", IntOrLongMatcher.isIntOrLong());
6863
MapMatcher.assertMap(EsqlTestUtils.singleValue(listResult.values()), basicMatcher);
6964

70-
request = new Request("GET", "/_query/queries/" + taskId + "1234");
71-
response = getRestClient().performRequest(request);
65+
Response getQueryResponse = getRestClient().performRequest(new Request("GET", "/_query/queries/" + taskId));
7266
MapMatcher.assertMap(
73-
entityToMap(response.getEntity()),
67+
jsonEntityToMap(getQueryResponse.getEntity()),
7468
basicMatcher.entry("coordinating_node", isA(String.class))
7569
.entry("data_nodes", allOf(isA(List.class), everyItem(isA(String.class))))
7670
);
@@ -86,21 +80,9 @@ public void testRunningQueries() throws Exception {
8680
}
8781
}
8882

89-
// FIXME(gal, NOCOMMIT) copy paste
9083
private EsqlQueryResponse sendAsyncQuery() {
9184
scriptPermits.drainPermits();
9285
scriptPermits.release(between(1, 5));
9386
return EsqlQueryRequestBuilder.newAsyncEsqlQueryRequestBuilder(client()).query(QUERY).execute().actionGet(60, TimeUnit.SECONDS);
9487
}
95-
96-
// FIXME(gal, NOCOMMIT) copy pasted from another place
97-
@SuppressWarnings("unchecked")
98-
private static <T> Map<String, T> entityToMap(HttpEntity entity) throws IOException {
99-
try (InputStream content = entity.getContent()) {
100-
XContentType xContentType = XContentType.fromMediaType(entity.getContentType().getValue());
101-
assertEquals(XContentType.JSON, xContentType);
102-
var map = XContentHelper.convertToMap(xContentType.xContent(), content, false);
103-
return (Map<String, T>) map;
104-
}
105-
}
10688
}

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -937,7 +937,12 @@ public enum Cap {
937937
/**
938938
* Make numberOfChannels consistent with layout in DefaultLayout by removing duplicated ChannelSet.
939939
*/
940-
MAKE_NUMBER_OF_CHANNELS_CONSISTENT_WITH_LAYOUT;
940+
MAKE_NUMBER_OF_CHANNELS_CONSISTENT_WITH_LAYOUT,
941+
942+
/**
943+
* Listing queries and getting information on a specific query.
944+
*/
945+
QUERY_MONITORING;
941946

942947
private final boolean enabled;
943948

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
import org.elasticsearch.action.ActionResponse;
1111
import org.elasticsearch.common.io.stream.StreamOutput;
12+
import org.elasticsearch.xcontent.ToXContentFragment;
1213
import org.elasticsearch.xcontent.ToXContentObject;
1314
import org.elasticsearch.xcontent.XContentBuilder;
1415

@@ -20,7 +21,7 @@ public class EsqlListQueriesResponse extends ActionResponse implements ToXConten
2021

2122
public record Query(org.elasticsearch.tasks.TaskId taskId, long startTimeMillis, long runningTimeNanos, String query)
2223
implements
23-
ToXContentObject {
24+
ToXContentFragment {
2425
@Override
2526
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
2627
builder.startObject(taskId.toString());

0 commit comments

Comments
 (0)