Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add UpdateRequest support to High Level Rest client #23266

Merged
merged 4 commits into from
Feb 22, 2017
Merged

Add UpdateRequest support to High Level Rest client #23266

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e101f40
Add UpdateRequest support to High Level Rest client
tlrx Feb 20, 2017
0edc183
Update after Luca review
tlrx Feb 21, 2017
ebf3a7b
catch Exception
tlrx Feb 22, 2017
ea5033c
Just wrap request converter
tlrx Feb 22, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 61 additions & 1 deletion client/rest-high-level/src/main/java/org/elasticsearch/client/Request.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,12 +32,16 @@
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.lucene.uid.Versions;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Locale;
Expand Down Expand Up @@ -118,6 +122,48 @@ static Request index(IndexRequest indexRequest) {
return new Request(method, endpoint, parameters.getParams(), entity);
}

static Request update(UpdateRequest updateRequest) throws IOException {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

wondering if we should rather catch IOException and rethrow UncheckedIOException. getting tired of all these checked exceptions, which force us to move to CheckedFunction.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, I'm on the fence on this. We already use CheckedFunction for response parsers I think we could also use CheckedFunction for request converters? I don't feel like it adds a lot of boling code.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok fine with me

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks

String endpoint = endpoint(updateRequest.index(), updateRequest.type(), updateRequest.id(), "_update");

Params parameters = Params.builder();
parameters.withRouting(updateRequest.routing());
parameters.withParent(updateRequest.parent());
parameters.withTimeout(updateRequest.timeout());
parameters.withRefreshPolicy(updateRequest.getRefreshPolicy());
parameters.withWaitForActiveShards(updateRequest.waitForActiveShards());
parameters.withDocAsUpsert(updateRequest.docAsUpsert());
parameters.withFetchSourceContext(updateRequest.fetchSource());
parameters.withRetryOnConflict(updateRequest.retryOnConflict());
parameters.withVersion(updateRequest.version());
parameters.withVersionType(updateRequest.versionType());

// The Java API allows update requests with different content types
// set for the partial document and the upsert document. This client
// only accepts update requests that have the same content types set
// for both doc and upsert.
XContentType xContentType = null;
if (updateRequest.doc() != null) {
xContentType = updateRequest.doc().getContentType();
}
if (updateRequest.upsertRequest() != null) {
XContentType upsertContentType = updateRequest.upsertRequest().getContentType();
if ((xContentType != null) && (xContentType != upsertContentType)) {
throw new IllegalStateException("Update request cannot have different content types for doc [" + xContentType + "]" +
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that the assumption up until now was that these methods wouldn't throw exceptions. if they do, we have to modify the async methods to catch and call the listener. I think users wouldn't expect to have exceptions thrown with async methods, even if this happens synchronously before the request gets sent :)

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if they do, we have to modify the async methods to catch and call the listener.

I think I did it in performRequestAsync() ?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oh I had missed this. But we only catch IOException, while we can also throw IllegalStateException now?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right - it should catch Exception instead (and also ensure that the listener get called anyway because I just notive that performRequestAsync throws an IllegalArgumentException too.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oh oh, I think those exceptions thrown in RestClient#performRequestAsync are a problem to be handled in the low level client. the bug is there I'm afraid.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oh right - I didn't notice I was looking at the RestClient itself. I'll take care of changing it

" and upsert [" + upsertContentType + "] documents");
} else {
xContentType = upsertContentType;
}
}
if (xContentType == null) {
xContentType = Requests.INDEX_CONTENT_TYPE;
}

BytesRef source = XContentHelper.toXContent(updateRequest, xContentType, false).toBytesRef();
HttpEntity entity = new ByteArrayEntity(source.bytes, source.offset, source.length, ContentType.create(xContentType.mediaType()));

return new Request(HttpPost.METHOD_NAME, endpoint, parameters.getParams(), entity);
}

/**
* Utility method to build request's endpoint.
*/
Expand Down Expand Up @@ -160,6 +206,13 @@ Params putParam(String key, TimeValue value) {
return this;
}

Params withDocAsUpsert(boolean docAsUpsert) {
if (docAsUpsert) {
return putParam("doc_as_upsert", Boolean.TRUE.toString());
}
return this;
}

Params withFetchSourceContext(FetchSourceContext fetchSourceContext) {
if (fetchSourceContext != null) {
if (fetchSourceContext.fetchSource() == false) {
Expand Down Expand Up @@ -203,7 +256,14 @@ Params withRefresh(boolean refresh) {

Params withRefreshPolicy(WriteRequest.RefreshPolicy refreshPolicy) {
if (refreshPolicy != WriteRequest.RefreshPolicy.NONE) {
putParam("refresh", refreshPolicy.getValue());
return putParam("refresh", refreshPolicy.getValue());
}
return this;
}

Params withRetryOnConflict(int retryOnConflict) {
if (retryOnConflict > 0) {
return putParam("retry_on_conflict", String.valueOf(retryOnConflict));
}
return this;
}
Expand Down
60 changes: 44 additions & 16 deletions client/rest-high-level/src/main/java/org/elasticsearch/client/RestHighLevelClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.main.MainRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.common.CheckedFunction;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.XContentParser;
Expand All @@ -39,10 +41,8 @@
import org.elasticsearch.rest.RestStatus;

import java.io.IOException;
import java.util.Collections;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;

import static java.util.Collections.emptySet;
import static java.util.Collections.singleton;
Expand Down Expand Up @@ -121,14 +121,35 @@ public void indexAsync(IndexRequest indexRequest, ActionListener<IndexResponse>
performRequestAsyncAndParseEntity(indexRequest, Request::index, IndexResponse::fromXContent, listener, emptySet(), headers);
}

private <Req extends ActionRequest, Resp> Resp performRequestAndParseEntity(Req request, Function<Req, Request> requestConverter,
CheckedFunction<XContentParser, Resp, IOException> entityParser, Set<Integer> ignores, Header... headers) throws IOException {
return performRequest(request, requestConverter, (response) -> parseEntity(response.getEntity(), entityParser), ignores, headers);
/**
* Updates a document using the Update API
* <p>
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html">Update API on elastic.co</a>
*/
public UpdateResponse update(UpdateRequest updateRequest, Header... headers) throws IOException {
return performRequestAndParseEntity(updateRequest, Request::update, UpdateResponse::fromXContent, emptySet(), headers);
}

<Req extends ActionRequest, Resp> Resp performRequest(Req request, Function<Req, Request> requestConverter,
CheckedFunction<Response, Resp, IOException> responseConverter, Set<Integer> ignores, Header... headers) throws IOException {
/**
* Asynchronously updates a document using the Update API
* <p>
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html">Update API on elastic.co</a>
*/
public void updateAsync(UpdateRequest updateRequest, ActionListener<UpdateResponse> listener, Header... headers) {
performRequestAsyncAndParseEntity(updateRequest, Request::update, UpdateResponse::fromXContent, listener, emptySet(), headers);
}

private <Req extends ActionRequest, Resp> Resp performRequestAndParseEntity(Req request,
CheckedFunction<Req, Request, IOException> requestConverter,
CheckedFunction<XContentParser, Resp, IOException> entityParser,
Set<Integer> ignores, Header... headers) throws IOException {
return performRequest(request, requestConverter, (response) -> parseEntity(response.getEntity(), entityParser), ignores, headers);
}

<Req extends ActionRequest, Resp> Resp performRequest(Req request,
CheckedFunction<Req, Request, IOException> requestConverter,
CheckedFunction<Response, Resp, IOException> responseConverter,
Set<Integer> ignores, Header... headers) throws IOException {
ActionRequestValidationException validationException = request.validate();
if (validationException != null) {
throw validationException;
Expand All @@ -154,24 +175,31 @@ <Req extends ActionRequest, Resp> Resp performRequest(Req request, Function<Req,
}
}

private <Req extends ActionRequest, Resp> void performRequestAsyncAndParseEntity(Req request, Function<Req, Request> requestConverter,
CheckedFunction<XContentParser, Resp, IOException> entityParser, ActionListener<Resp> listener,
Set<Integer> ignores, Header... headers) {
private <Req extends ActionRequest, Resp> void performRequestAsyncAndParseEntity(Req request,
CheckedFunction<Req, Request, IOException> requestConverter,
CheckedFunction<XContentParser, Resp, IOException> entityParser,
ActionListener<Resp> listener, Set<Integer> ignores, Header... headers) {
performRequestAsync(request, requestConverter, (response) -> parseEntity(response.getEntity(), entityParser),
listener, ignores, headers);
}

<Req extends ActionRequest, Resp> void performRequestAsync(Req request, Function<Req, Request> requestConverter,
CheckedFunction<Response, Resp, IOException> responseConverter, ActionListener<Resp> listener,
Set<Integer> ignores, Header... headers) {
<Req extends ActionRequest, Resp> void performRequestAsync(Req request,
CheckedFunction<Req, Request, IOException> requestConverter,
CheckedFunction<Response, Resp, IOException> responseConverter,
ActionListener<Resp> listener, Set<Integer> ignores, Header... headers) {
ActionRequestValidationException validationException = request.validate();
if (validationException != null) {
listener.onFailure(validationException);
return;
}
Request req = requestConverter.apply(request);
ResponseListener responseListener = wrapResponseListener(responseConverter, listener, ignores);
client.performRequestAsync(req.method, req.endpoint, req.params, req.entity, responseListener, headers);
try {
Request req = requestConverter.apply(request);

ResponseListener responseListener = wrapResponseListener(responseConverter, listener, ignores);
client.performRequestAsync(req.method, req.endpoint, req.params, req.entity, responseListener, headers);
} catch (Exception e) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure here, I think for correctness we should only wrap the requestConverter call? the wrapResponseListener and performRequestAsync should really never throw exception. WDYT?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that what I did first but then I wrapped everything because I mixed up the performRequestAsync methods.

listener.onFailure(e);
}
}

static <Resp> ResponseListener wrapResponseListener(CheckedFunction<Response, Resp, IOException> responseConverter,
Expand Down
Loading