HLRC: Add throttling for update & delete-by-query (#33951)

This change adds throttling to the update-by-query and delete-by-query cases
similar to throttling for reindex. This mostly means additional methods on the
client class itself, since the request hits the same RestHandler, just with
slightly different endpoints, and also the return values are similar.
This commit is contained in:
Christoph Büscher 2018-10-02 21:44:15 +02:00 committed by GitHub
parent dd3fe92673
commit a1c441f78a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 288 additions and 47 deletions

View file

@ -530,8 +530,20 @@ final class RequestConverters {
return request; return request;
} }
static Request rethrottle(RethrottleRequest rethrottleRequest) throws IOException { static Request rethrottleReindex(RethrottleRequest rethrottleRequest) {
String endpoint = new EndpointBuilder().addPathPart("_reindex").addPathPart(rethrottleRequest.getTaskId().toString()) return rethrottle(rethrottleRequest, "_reindex");
}
static Request rethrottleUpdateByQuery(RethrottleRequest rethrottleRequest) {
return rethrottle(rethrottleRequest, "_update_by_query");
}
static Request rethrottleDeleteByQuery(RethrottleRequest rethrottleRequest) {
return rethrottle(rethrottleRequest, "_delete_by_query");
}
private static Request rethrottle(RethrottleRequest rethrottleRequest, String firstPathPart) {
String endpoint = new EndpointBuilder().addPathPart(firstPathPart).addPathPart(rethrottleRequest.getTaskId().toString())
.addPathPart("_rethrottle").build(); .addPathPart("_rethrottle").build();
Request request = new Request(HttpPost.METHOD_NAME, endpoint); Request request = new Request(HttpPost.METHOD_NAME, endpoint);
Params params = new Params(request) Params params = new Params(request)

View file

@ -516,6 +516,62 @@ public class RestHighLevelClient implements Closeable {
); );
} }
/**
* Executes a delete by query rethrottle request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-delete-by-query.html">
* Delete By Query API on elastic.co</a>
* @param rethrottleRequest the request
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @return the response
* @throws IOException in case there is a problem sending the request or parsing back the response
*/
public final ListTasksResponse deleteByQueryRethrottle(RethrottleRequest rethrottleRequest, RequestOptions options) throws IOException {
return performRequestAndParseEntity(rethrottleRequest, RequestConverters::rethrottleDeleteByQuery, options,
ListTasksResponse::fromXContent, emptySet());
}
/**
* Asynchronously execute an delete by query rethrottle request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-delete-by-query.html">
* Delete By Query API on elastic.co</a>
* @param rethrottleRequest the request
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @param listener the listener to be notified upon request completion
*/
public final void deleteByQueryRethrottleAsync(RethrottleRequest rethrottleRequest, RequestOptions options,
ActionListener<ListTasksResponse> listener) {
performRequestAsyncAndParseEntity(rethrottleRequest, RequestConverters::rethrottleDeleteByQuery, options,
ListTasksResponse::fromXContent, listener, emptySet());
}
/**
* Executes a update by query rethrottle request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html">
* Update By Query API on elastic.co</a>
* @param rethrottleRequest the request
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @return the response
* @throws IOException in case there is a problem sending the request or parsing back the response
*/
public final ListTasksResponse updateByQueryRethrottle(RethrottleRequest rethrottleRequest, RequestOptions options) throws IOException {
return performRequestAndParseEntity(rethrottleRequest, RequestConverters::rethrottleUpdateByQuery, options,
ListTasksResponse::fromXContent, emptySet());
}
/**
* Asynchronously execute an update by query rethrottle request.
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html">
* Update By Query API on elastic.co</a>
* @param rethrottleRequest the request
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
* @param listener the listener to be notified upon request completion
*/
public final void updateByQueryRethrottleAsync(RethrottleRequest rethrottleRequest, RequestOptions options,
ActionListener<ListTasksResponse> listener) {
performRequestAsyncAndParseEntity(rethrottleRequest, RequestConverters::rethrottleUpdateByQuery, options,
ListTasksResponse::fromXContent, listener, emptySet());
}
/** /**
* Executes a reindex rethrottling request. * Executes a reindex rethrottling request.
* See the <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-reindex.html#docs-reindex-rethrottle"> * See the <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-reindex.html#docs-reindex-rethrottle">
@ -527,8 +583,8 @@ public class RestHighLevelClient implements Closeable {
* @throws IOException in case there is a problem sending the request or parsing back the response * @throws IOException in case there is a problem sending the request or parsing back the response
*/ */
public final ListTasksResponse reindexRethrottle(RethrottleRequest rethrottleRequest, RequestOptions options) throws IOException { public final ListTasksResponse reindexRethrottle(RethrottleRequest rethrottleRequest, RequestOptions options) throws IOException {
return performRequestAndParseEntity(rethrottleRequest, RequestConverters::rethrottle, options, ListTasksResponse::fromXContent, return performRequestAndParseEntity(rethrottleRequest, RequestConverters::rethrottleReindex, options,
emptySet()); ListTasksResponse::fromXContent, emptySet());
} }
/** /**
@ -541,9 +597,9 @@ public class RestHighLevelClient implements Closeable {
* @param listener the listener to be notified upon request completion * @param listener the listener to be notified upon request completion
*/ */
public final void reindexRethrottleAsync(RethrottleRequest rethrottleRequest, RequestOptions options, public final void reindexRethrottleAsync(RethrottleRequest rethrottleRequest, RequestOptions options,
ActionListener<ListTasksResponse> listener) { ActionListener<ListTasksResponse> listener) {
performRequestAsyncAndParseEntity(rethrottleRequest, RequestConverters::rethrottle, options, ListTasksResponse::fromXContent, performRequestAsyncAndParseEntity(rethrottleRequest, RequestConverters::rethrottleReindex, options, ListTasksResponse::fromXContent,
listener, emptySet()); listener, emptySet());
} }
/** /**

View file

@ -55,9 +55,11 @@ import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.get.GetResult; import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.query.IdsQueryBuilder; import org.elasticsearch.index.query.IdsQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse; import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.index.reindex.DeleteByQueryRequest; import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.ReindexAction; import org.elasticsearch.index.reindex.ReindexAction;
import org.elasticsearch.index.reindex.ReindexRequest; import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.UpdateByQueryAction;
import org.elasticsearch.index.reindex.UpdateByQueryRequest; import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus; import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script; import org.elasticsearch.script.Script;
@ -727,10 +729,7 @@ public class CrudIT extends ESRestHighLevelClientTestCase {
} }
}); });
TaskGroup taskGroupToRethrottle = findTaskToRethrottle(); TaskId taskIdToRethrottle = findTaskToRethrottle(ReindexAction.NAME);
assertThat(taskGroupToRethrottle.getChildTasks(), empty());
TaskId taskIdToRethrottle = taskGroupToRethrottle.getTaskInfo().getTaskId();
float requestsPerSecond = 1000f; float requestsPerSecond = 1000f;
ListTasksResponse response = execute(new RethrottleRequest(taskIdToRethrottle, requestsPerSecond), ListTasksResponse response = execute(new RethrottleRequest(taskIdToRethrottle, requestsPerSecond),
highLevelClient()::reindexRethrottle, highLevelClient()::reindexRethrottleAsync); highLevelClient()::reindexRethrottle, highLevelClient()::reindexRethrottleAsync);
@ -752,10 +751,10 @@ public class CrudIT extends ESRestHighLevelClientTestCase {
} }
} }
private TaskGroup findTaskToRethrottle() throws IOException { private TaskId findTaskToRethrottle(String actionName) throws IOException {
long start = System.nanoTime(); long start = System.nanoTime();
ListTasksRequest request = new ListTasksRequest(); ListTasksRequest request = new ListTasksRequest();
request.setActions(ReindexAction.NAME); request.setActions(actionName);
request.setDetailed(true); request.setDetailed(true);
do { do {
ListTasksResponse list = highLevelClient().tasks().list(request, RequestOptions.DEFAULT); ListTasksResponse list = highLevelClient().tasks().list(request, RequestOptions.DEFAULT);
@ -766,13 +765,15 @@ public class CrudIT extends ESRestHighLevelClientTestCase {
// The parent task hasn't started yet // The parent task hasn't started yet
continue; continue;
} }
return list.getTaskGroups().get(0); TaskGroup taskGroup = list.getTaskGroups().get(0);
assertThat(taskGroup.getChildTasks(), empty());
return taskGroup.getTaskInfo().getTaskId();
} while (System.nanoTime() - start < TimeUnit.SECONDS.toNanos(10)); } while (System.nanoTime() - start < TimeUnit.SECONDS.toNanos(10));
throw new AssertionError("Couldn't find tasks to rethrottle. Here are the running tasks " + throw new AssertionError("Couldn't find tasks to rethrottle. Here are the running tasks " +
highLevelClient().tasks().list(request, RequestOptions.DEFAULT)); highLevelClient().tasks().list(request, RequestOptions.DEFAULT));
} }
public void testUpdateByQuery() throws IOException { public void testUpdateByQuery() throws Exception {
final String sourceIndex = "source1"; final String sourceIndex = "source1";
{ {
// Prepare // Prepare
@ -836,9 +837,53 @@ public class CrudIT extends ESRestHighLevelClientTestCase {
.getSourceAsMap().get("foo")) .getSourceAsMap().get("foo"))
); );
} }
{
// test update-by-query rethrottling
UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest();
updateByQueryRequest.indices(sourceIndex);
updateByQueryRequest.setQuery(new IdsQueryBuilder().addIds("1").types("type"));
updateByQueryRequest.setRefresh(true);
// this following settings are supposed to halt reindexing after first document
updateByQueryRequest.setBatchSize(1);
updateByQueryRequest.setRequestsPerSecond(0.00001f);
final CountDownLatch taskFinished = new CountDownLatch(1);
highLevelClient().updateByQueryAsync(updateByQueryRequest, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {
@Override
public void onResponse(BulkByScrollResponse response) {
taskFinished.countDown();
}
@Override
public void onFailure(Exception e) {
fail(e.toString());
}
});
TaskId taskIdToRethrottle = findTaskToRethrottle(UpdateByQueryAction.NAME);
float requestsPerSecond = 1000f;
ListTasksResponse response = execute(new RethrottleRequest(taskIdToRethrottle, requestsPerSecond),
highLevelClient()::updateByQueryRethrottle, highLevelClient()::updateByQueryRethrottleAsync);
assertThat(response.getTasks(), hasSize(1));
assertEquals(taskIdToRethrottle, response.getTasks().get(0).getTaskId());
assertThat(response.getTasks().get(0).getStatus(), instanceOf(RawTaskStatus.class));
assertEquals(Float.toString(requestsPerSecond),
((RawTaskStatus) response.getTasks().get(0).getStatus()).toMap().get("requests_per_second").toString());
taskFinished.await(2, TimeUnit.SECONDS);
// any rethrottling after the update-by-query is done performed with the same taskId should result in a failure
response = execute(new RethrottleRequest(taskIdToRethrottle, requestsPerSecond),
highLevelClient()::updateByQueryRethrottle, highLevelClient()::updateByQueryRethrottleAsync);
assertTrue(response.getTasks().isEmpty());
assertFalse(response.getNodeFailures().isEmpty());
assertEquals(1, response.getNodeFailures().size());
assertEquals("Elasticsearch exception [type=resource_not_found_exception, reason=task [" + taskIdToRethrottle + "] is missing]",
response.getNodeFailures().get(0).getCause().getMessage());
}
} }
public void testDeleteByQuery() throws IOException { public void testDeleteByQuery() throws Exception {
final String sourceIndex = "source1"; final String sourceIndex = "source1";
{ {
// Prepare // Prepare
@ -855,6 +900,8 @@ public class CrudIT extends ESRestHighLevelClientTestCase {
.source(Collections.singletonMap("foo", 1), XContentType.JSON)) .source(Collections.singletonMap("foo", 1), XContentType.JSON))
.add(new IndexRequest(sourceIndex, "type", "2") .add(new IndexRequest(sourceIndex, "type", "2")
.source(Collections.singletonMap("foo", 2), XContentType.JSON)) .source(Collections.singletonMap("foo", 2), XContentType.JSON))
.add(new IndexRequest(sourceIndex, "type", "3")
.source(Collections.singletonMap("foo", 3), XContentType.JSON))
.setRefreshPolicy(RefreshPolicy.IMMEDIATE), .setRefreshPolicy(RefreshPolicy.IMMEDIATE),
RequestOptions.DEFAULT RequestOptions.DEFAULT
).status() ).status()
@ -878,10 +925,54 @@ public class CrudIT extends ESRestHighLevelClientTestCase {
assertEquals(0, bulkResponse.getBulkFailures().size()); assertEquals(0, bulkResponse.getBulkFailures().size());
assertEquals(0, bulkResponse.getSearchFailures().size()); assertEquals(0, bulkResponse.getSearchFailures().size());
assertEquals( assertEquals(
1, 2,
highLevelClient().search(new SearchRequest(sourceIndex), RequestOptions.DEFAULT).getHits().totalHits highLevelClient().search(new SearchRequest(sourceIndex), RequestOptions.DEFAULT).getHits().totalHits
); );
} }
{
// test delete-by-query rethrottling
DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest();
deleteByQueryRequest.indices(sourceIndex);
deleteByQueryRequest.setQuery(new IdsQueryBuilder().addIds("2", "3").types("type"));
deleteByQueryRequest.setRefresh(true);
// this following settings are supposed to halt reindexing after first document
deleteByQueryRequest.setBatchSize(1);
deleteByQueryRequest.setRequestsPerSecond(0.00001f);
final CountDownLatch taskFinished = new CountDownLatch(1);
highLevelClient().deleteByQueryAsync(deleteByQueryRequest, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {
@Override
public void onResponse(BulkByScrollResponse response) {
taskFinished.countDown();
}
@Override
public void onFailure(Exception e) {
fail(e.toString());
}
});
TaskId taskIdToRethrottle = findTaskToRethrottle(DeleteByQueryAction.NAME);
float requestsPerSecond = 1000f;
ListTasksResponse response = execute(new RethrottleRequest(taskIdToRethrottle, requestsPerSecond),
highLevelClient()::deleteByQueryRethrottle, highLevelClient()::deleteByQueryRethrottleAsync);
assertThat(response.getTasks(), hasSize(1));
assertEquals(taskIdToRethrottle, response.getTasks().get(0).getTaskId());
assertThat(response.getTasks().get(0).getStatus(), instanceOf(RawTaskStatus.class));
assertEquals(Float.toString(requestsPerSecond),
((RawTaskStatus) response.getTasks().get(0).getStatus()).toMap().get("requests_per_second").toString());
taskFinished.await(2, TimeUnit.SECONDS);
// any rethrottling after the delete-by-query is done performed with the same taskId should result in a failure
response = execute(new RethrottleRequest(taskIdToRethrottle, requestsPerSecond),
highLevelClient()::deleteByQueryRethrottle, highLevelClient()::deleteByQueryRethrottleAsync);
assertTrue(response.getTasks().isEmpty());
assertFalse(response.getNodeFailures().isEmpty());
assertEquals(1, response.getNodeFailures().size());
assertEquals("Elasticsearch exception [type=resource_not_found_exception, reason=task [" + taskIdToRethrottle + "] is missing]",
response.getNodeFailures().get(0).getCause().getMessage());
}
} }
public void testBulkProcessorIntegration() throws IOException { public void testBulkProcessorIntegration() throws IOException {

View file

@ -59,6 +59,7 @@ import org.elasticsearch.common.CheckedBiConsumer;
import org.elasticsearch.common.Strings; import org.elasticsearch.common.Strings;
import org.elasticsearch.common.bytes.BytesArray; import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.Streams; import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.lucene.uid.Versions; import org.elasticsearch.common.lucene.uid.Versions;
import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.unit.TimeValue;
@ -466,7 +467,7 @@ public class RequestConvertersTests extends ESTestCase {
assertToXContentBody(deleteByQueryRequest, request.getEntity()); assertToXContentBody(deleteByQueryRequest, request.getEntity());
} }
public void testRethrottle() throws IOException { public void testRethrottle() {
TaskId taskId = new TaskId(randomAlphaOfLength(10), randomIntBetween(1, 100)); TaskId taskId = new TaskId(randomAlphaOfLength(10), randomIntBetween(1, 100));
RethrottleRequest rethrottleRequest; RethrottleRequest rethrottleRequest;
Float requestsPerSecond; Float requestsPerSecond;
@ -480,11 +481,20 @@ public class RequestConvertersTests extends ESTestCase {
expectedParams.put(RethrottleRequest.REQUEST_PER_SECOND_PARAMETER, "-1"); expectedParams.put(RethrottleRequest.REQUEST_PER_SECOND_PARAMETER, "-1");
} }
expectedParams.put("group_by", "none"); expectedParams.put("group_by", "none");
Request request = RequestConverters.rethrottle(rethrottleRequest); List<Tuple<String, Supplier<Request>>> variants = new ArrayList<>();
assertEquals("/_reindex/" + taskId + "/_rethrottle", request.getEndpoint()); variants.add(new Tuple<String, Supplier<Request>>("_reindex", () -> RequestConverters.rethrottleReindex(rethrottleRequest)));
assertEquals(HttpPost.METHOD_NAME, request.getMethod()); variants.add(new Tuple<String, Supplier<Request>>("_update_by_query",
assertEquals(expectedParams, request.getParameters()); () -> RequestConverters.rethrottleUpdateByQuery(rethrottleRequest)));
assertNull(request.getEntity()); variants.add(new Tuple<String, Supplier<Request>>("_delete_by_query",
() -> RequestConverters.rethrottleDeleteByQuery(rethrottleRequest)));
for (Tuple<String, Supplier<Request>> variant : variants) {
Request request = variant.v2().get();
assertEquals("/" + variant.v1() + "/" + taskId + "/_rethrottle", request.getEndpoint());
assertEquals(HttpPost.METHOD_NAME, request.getMethod());
assertEquals(expectedParams, request.getParameters());
assertNull(request.getEntity());
}
// test illegal RethrottleRequest values // test illegal RethrottleRequest values
Exception e = expectThrows(NullPointerException.class, () -> new RethrottleRequest(null, 1.0f)); Exception e = expectThrows(NullPointerException.class, () -> new RethrottleRequest(null, 1.0f));

View file

@ -902,19 +902,26 @@ public class CRUDDocumentationIT extends ESRestHighLevelClientTestCase {
TaskId taskId = new TaskId("oTUltX4IQMOUUVeiohTt8A:124"); TaskId taskId = new TaskId("oTUltX4IQMOUUVeiohTt8A:124");
{ {
// tag::rethrottle-disable-request // tag::rethrottle-disable-request
RethrottleRequest rethrottleRequest = new RethrottleRequest(taskId); // <1> RethrottleRequest request = new RethrottleRequest(taskId); // <1>
client.reindexRethrottle(rethrottleRequest, RequestOptions.DEFAULT);
// end::rethrottle-disable-request // end::rethrottle-disable-request
} }
{ {
// tag::rethrottle-request // tag::rethrottle-request
RethrottleRequest rethrottleRequest = new RethrottleRequest(taskId, 100.0f); // <1> RethrottleRequest request = new RethrottleRequest(taskId, 100.0f); // <1>
client.reindexRethrottle(rethrottleRequest, RequestOptions.DEFAULT);
// end::rethrottle-request // end::rethrottle-request
} }
// tag::rethrottle-request-async {
RethrottleRequest request = new RethrottleRequest(taskId);
// tag::rethrottle-request-execution
client.reindexRethrottle(request, RequestOptions.DEFAULT); // <1>
client.updateByQueryRethrottle(request, RequestOptions.DEFAULT); // <2>
client.deleteByQueryRethrottle(request, RequestOptions.DEFAULT); // <3>
// end::rethrottle-request-execution
}
// tag::rethrottle-request-async-listener
ActionListener<ListTasksResponse> listener = new ActionListener<ListTasksResponse>() { ActionListener<ListTasksResponse> listener = new ActionListener<ListTasksResponse>() {
@Override @Override
public void onResponse(ListTasksResponse response) { public void onResponse(ListTasksResponse response) {
@ -926,15 +933,17 @@ public class CRUDDocumentationIT extends ESRestHighLevelClientTestCase {
// <2> // <2>
} }
}; };
// end::rethrottle-request-async // end::rethrottle-request-async-listener
// Replace the empty listener by a blocking listener in test // Replace the empty listener by a blocking listener in test
final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch latch = new CountDownLatch(3);
listener = new LatchedActionListener<>(listener, latch); listener = new LatchedActionListener<>(listener, latch);
RethrottleRequest rethrottleRequest = new RethrottleRequest(taskId); RethrottleRequest request = new RethrottleRequest(taskId);
// tag::rethrottle-execute-async // tag::rethrottle-execute-async
client.reindexRethrottleAsync(rethrottleRequest, RequestOptions.DEFAULT, listener); // <1> client.reindexRethrottleAsync(request, RequestOptions.DEFAULT, listener); // <1>
client.updateByQueryRethrottleAsync(request, RequestOptions.DEFAULT, listener); // <2>
client.deleteByQueryRethrottleAsync(request, RequestOptions.DEFAULT, listener); // <3>
// end::rethrottle-execute-async // end::rethrottle-execute-async
assertTrue(latch.await(30L, TimeUnit.SECONDS)); assertTrue(latch.await(30L, TimeUnit.SECONDS));
} }

View file

@ -1,15 +1,15 @@
[[java-rest-high-document-reindex-rethrottle]] [[java-rest-high-document-rethrottle]]
=== Reindex Rethrottle API === Rethrottle API
[[java-rest-high-document-reindex-rethrottle-request]] [[java-rest-high-document-rethrottle-request]]
==== Reindex Rethrolle Request ==== Rethrottle Request
A `RethrottleRequest` can be used to change existing throttling on a runnind A `RethrottleRequest` can be used to change the current throttling on a running
reindex task or disable it entirely. It requires the task Id of the reindex reindex, update-by-query or delete-by-query task or to disable throttling of
task to change. the task entirely. It requires the task Id of the task to change.
In its simplest form, you can use it to disable throttling of a running In its simplest form, you can use it to disable throttling of a running
reindex task using the following: task using the following:
["source","java",subs="attributes,callouts,macros"] ["source","java",subs="attributes,callouts,macros"]
-------------------------------------------------- --------------------------------------------------
@ -26,7 +26,19 @@ include-tagged::{doc-tests}/CRUDDocumentationIT.java[rethrottle-request]
-------------------------------------------------- --------------------------------------------------
<1> Request to change the throttling of a task to 100 requests per second <1> Request to change the throttling of a task to 100 requests per second
[[java-rest-high-document-reindex-rethrottle-async]] The rethrottling request can be executed by using one of the three appropriate
methods depending on whether a reindex, update-by-query or delete-by-query task
should be rethrottled:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests}/CRUDDocumentationIT.java[rethrottle-request-execution]
--------------------------------------------------
<1> Execute reindex rethrottling request
<2> The same for update-by-query
<3> The same for delete-by-query
[[java-rest-high-document-rethrottle-async]]
==== Asynchronous Execution ==== Asynchronous Execution
The asynchronous execution of a rethrottle request requires both the `RethrottleRequest` The asynchronous execution of a rethrottle request requires both the `RethrottleRequest`
@ -37,8 +49,9 @@ method:
-------------------------------------------------- --------------------------------------------------
include-tagged::{doc-tests}/CRUDDocumentationIT.java[rethrottle-execute-async] include-tagged::{doc-tests}/CRUDDocumentationIT.java[rethrottle-execute-async]
-------------------------------------------------- --------------------------------------------------
<1> The RethrottleRequest to execute and the ActionListener to use when the <1> Execute reindex rethrottling asynchronously
execution completes <2> The same for update-by-query
<3> The same for delete-by-query
The asynchronous method does not block and returns immediately. The asynchronous method does not block and returns immediately.
Once it is completed the `ActionListener` is called back using the `onResponse` method Once it is completed the `ActionListener` is called back using the `onResponse` method
@ -47,12 +60,12 @@ it failed. A typical listener looks like this:
["source","java",subs="attributes,callouts,macros"] ["source","java",subs="attributes,callouts,macros"]
-------------------------------------------------- --------------------------------------------------
include-tagged::{doc-tests}/CRUDDocumentationIT.java[rethrottle-request-async] include-tagged::{doc-tests}/CRUDDocumentationIT.java[rethrottle-request-async-listener]
-------------------------------------------------- --------------------------------------------------
<1> Code executed when the request is successfully completed <1> Code executed when the request is successfully completed
<2> Code executed when the request fails with an exception <2> Code executed when the request fails with an exception
[[java-rest-high-document-reindex-retrottle-response]] [[java-rest-high-document-retrottle-response]]
==== Rethrottle Response ==== Rethrottle Response
Rethrottling returns the task that has been rethrottled in the form of a Rethrottling returns the task that has been rethrottled in the form of a

View file

@ -33,7 +33,7 @@ include::document/multi-get.asciidoc[]
include::document/reindex.asciidoc[] include::document/reindex.asciidoc[]
include::document/update-by-query.asciidoc[] include::document/update-by-query.asciidoc[]
include::document/delete-by-query.asciidoc[] include::document/delete-by-query.asciidoc[]
include::document/reindex-rethrottle.asciidoc[] include::document/rethrottle.asciidoc[]
== Search APIs == Search APIs

View file

@ -0,0 +1,25 @@
{
"delete_by_query_rethrottle": {
"documentation": "https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-delete-by-query.html",
"methods": ["POST"],
"url": {
"path": "/_delete_by_query/{task_id}/_rethrottle",
"paths": ["/_delete_by_query/{task_id}/_rethrottle"],
"parts": {
"task_id": {
"type": "string",
"required" : true,
"description": "The task id to rethrottle"
}
},
"params": {
"requests_per_second": {
"type": "number",
"required": true,
"description": "The throttle to set on this request in floating sub-requests per second. -1 means set no throttle."
}
}
},
"body": null
}
}

View file

@ -4,7 +4,7 @@
"methods": ["POST"], "methods": ["POST"],
"url": { "url": {
"path": "/_reindex/{task_id}/_rethrottle", "path": "/_reindex/{task_id}/_rethrottle",
"paths": ["/_reindex/{task_id}/_rethrottle", "/_update_by_query/{task_id}/_rethrottle", "/_delete_by_query/{task_id}/_rethrottle"], "paths": ["/_reindex/{task_id}/_rethrottle"],
"parts": { "parts": {
"task_id": { "task_id": {
"type": "string", "type": "string",

View file

@ -0,0 +1,25 @@
{
"update_by_query_rethrottle": {
"documentation": "https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update-by-query.html",
"methods": ["POST"],
"url": {
"path": "/_update_by_query/{task_id}/_rethrottle",
"paths": ["/_update_by_query/{task_id}/_rethrottle"],
"parts": {
"task_id": {
"type": "string",
"required" : true,
"description": "The task id to rethrottle"
}
},
"params": {
"requests_per_second": {
"type": "number",
"required": true,
"description": "The throttle to set on this request in floating sub-requests per second. -1 means set no throttle."
}
}
},
"body": null
}
}