diff --git a/client/rest/src/main/java/org/elasticsearch/client/RequestLogger.java b/client/rest/src/main/java/org/elasticsearch/client/RequestLogger.java index 24e6881fa1e..fd4c3600234 100644 --- a/client/rest/src/main/java/org/elasticsearch/client/RequestLogger.java +++ b/client/rest/src/main/java/org/elasticsearch/client/RequestLogger.java @@ -81,7 +81,7 @@ final class RequestLogger { /** * Logs a request that failed */ - static void logFailedRequest(Log logger, HttpUriRequest request, HttpHost host, IOException e) { + static void logFailedRequest(Log logger, HttpUriRequest request, HttpHost host, Exception e) { if (logger.isDebugEnabled()) { logger.debug("request [" + request.getMethod() + " " + host + getUri(request.getRequestLine()) + "] failed", e); } diff --git a/client/rest/src/main/java/org/elasticsearch/client/Response.java b/client/rest/src/main/java/org/elasticsearch/client/Response.java index f7685b27bb9..be2a16f912d 100644 --- a/client/rest/src/main/java/org/elasticsearch/client/Response.java +++ b/client/rest/src/main/java/org/elasticsearch/client/Response.java @@ -22,12 +22,11 @@ package org.elasticsearch.client; import org.apache.http.Header; import org.apache.http.HttpEntity; import org.apache.http.HttpHost; +import org.apache.http.HttpResponse; import org.apache.http.RequestLine; import org.apache.http.StatusLine; import org.apache.http.client.methods.CloseableHttpResponse; -import java.io.Closeable; -import java.io.IOException; import java.util.Objects; /** @@ -35,13 +34,13 @@ import java.util.Objects; * its corresponding {@link RequestLine} and {@link HttpHost}. * It must be closed to free any resource held by it, as well as the corresponding connection in the connection pool. */ -public class Response implements Closeable { +public class Response { private final RequestLine requestLine; private final HttpHost host; - private final CloseableHttpResponse response; + private final HttpResponse response; - Response(RequestLine requestLine, HttpHost host, CloseableHttpResponse response) { + Response(RequestLine requestLine, HttpHost host, HttpResponse response) { Objects.requireNonNull(requestLine, "requestLine cannot be null"); Objects.requireNonNull(host, "node cannot be null"); Objects.requireNonNull(response, "response cannot be null"); @@ -107,9 +106,4 @@ public class Response implements Closeable { ", response=" + response.getStatusLine() + '}'; } - - @Override - public void close() throws IOException { - this.response.close(); - } } diff --git a/client/rest/src/main/java/org/elasticsearch/client/ResponseException.java b/client/rest/src/main/java/org/elasticsearch/client/ResponseException.java index 44f59cce7db..5b6c4f1f0e7 100644 --- a/client/rest/src/main/java/org/elasticsearch/client/ResponseException.java +++ b/client/rest/src/main/java/org/elasticsearch/client/ResponseException.java @@ -23,44 +23,26 @@ import java.io.IOException; /** * Exception thrown when an elasticsearch node responds to a request with a status code that indicates an error. - * Note that the response body gets passed in as a string and read eagerly, which means that the Response object - * is expected to be closed and available only to read metadata like status line, request line, response headers. + * Holds the response that was returned. */ public class ResponseException extends IOException { private Response response; - private final String responseBody; - ResponseException(Response response, String responseBody) throws IOException { - super(buildMessage(response,responseBody)); + ResponseException(Response response) throws IOException { + super(buildMessage(response)); this.response = response; - this.responseBody = responseBody; } - private static String buildMessage(Response response, String responseBody) { - String message = response.getRequestLine().getMethod() + " " + response.getHost() + response.getRequestLine().getUri() + private static String buildMessage(Response response) { + return response.getRequestLine().getMethod() + " " + response.getHost() + response.getRequestLine().getUri() + ": " + response.getStatusLine().toString(); - if (responseBody != null) { - message += "\n" + responseBody; - } - return message; } /** * Returns the {@link Response} that caused this exception to be thrown. - * Expected to be used only to read metadata like status line, request line, response headers. The response body should - * be retrieved using {@link #getResponseBody()} */ public Response getResponse() { return response; } - - /** - * Returns the response body as a string or null if there wasn't any. - * The body is eagerly consumed when an ResponseException gets created, and its corresponding Response - * gets closed straightaway so this method is the only way to get back the response body that was returned. - */ - public String getResponseBody() { - return responseBody; - } } diff --git a/client/rest/src/test/java/org/elasticsearch/client/CloseableBasicHttpResponse.java b/client/rest/src/main/java/org/elasticsearch/client/ResponseListener.java similarity index 51% rename from client/rest/src/test/java/org/elasticsearch/client/CloseableBasicHttpResponse.java rename to client/rest/src/main/java/org/elasticsearch/client/ResponseListener.java index dd866bac541..ce948f6569b 100644 --- a/client/rest/src/test/java/org/elasticsearch/client/CloseableBasicHttpResponse.java +++ b/client/rest/src/main/java/org/elasticsearch/client/ResponseListener.java @@ -19,24 +19,22 @@ package org.elasticsearch.client; -import org.apache.http.StatusLine; -import org.apache.http.client.methods.CloseableHttpResponse; -import org.apache.http.message.BasicHttpResponse; - -import java.io.IOException; - /** - * Simple {@link CloseableHttpResponse} impl needed to easily create http responses that are closeable given that - * org.apache.http.impl.execchain.HttpResponseProxy is not public. + * Listener to be provided when calling async performRequest methods provided by {@link RestClient}. + * Those methods that do accept a listener will return immediately, execute asynchronously, and notify + * the listener whenever the request yielded a response, or failed with an exception. */ -class CloseableBasicHttpResponse extends BasicHttpResponse implements CloseableHttpResponse { +public interface ResponseListener { - public CloseableBasicHttpResponse(StatusLine statusline) { - super(statusline); - } + /** + * Method invoked if the request yielded a successful response + */ + void onSuccess(Response response); - @Override - public void close() throws IOException { - //nothing to close - } -} \ No newline at end of file + /** + * Method invoked if the request failed. There are two main categories of failures: connection failures (usually + * {@link java.io.IOException}s, or responses that were treated as errors based on their error response code + * ({@link ResponseException}s). + */ + void onFailure(Exception exception); +} diff --git a/client/rest/src/main/java/org/elasticsearch/client/RestClient.java b/client/rest/src/main/java/org/elasticsearch/client/RestClient.java index 0d6941c7857..b5bfd6ee07c 100644 --- a/client/rest/src/main/java/org/elasticsearch/client/RestClient.java +++ b/client/rest/src/main/java/org/elasticsearch/client/RestClient.java @@ -25,9 +25,9 @@ import org.apache.http.Header; import org.apache.http.HttpEntity; import org.apache.http.HttpHost; import org.apache.http.HttpRequest; +import org.apache.http.HttpResponse; import org.apache.http.client.ClientProtocolException; import org.apache.http.client.config.RequestConfig; -import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpEntityEnclosingRequestBase; import org.apache.http.client.methods.HttpHead; import org.apache.http.client.methods.HttpOptions; @@ -37,11 +37,16 @@ import org.apache.http.client.methods.HttpPut; import org.apache.http.client.methods.HttpRequestBase; import org.apache.http.client.methods.HttpTrace; import org.apache.http.client.utils.URIBuilder; +import org.apache.http.concurrent.FutureCallback; import org.apache.http.entity.ContentType; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; -import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; -import org.apache.http.util.EntityUtils; +import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; +import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; +import org.apache.http.nio.client.methods.HttpAsyncMethods; +import org.apache.http.nio.conn.SchemeIOSessionStrategy; +import org.apache.http.nio.protocol.HttpAsyncRequestProducer; +import org.apache.http.nio.protocol.HttpAsyncResponseConsumer; import java.io.Closeable; import java.io.IOException; @@ -51,6 +56,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Locale; import java.util.Map; @@ -58,6 +64,7 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -79,9 +86,8 @@ public final class RestClient implements Closeable { private static final Log logger = LogFactory.getLog(RestClient.class); public static ContentType JSON_CONTENT_TYPE = ContentType.create("application/json", Consts.UTF_8); - private final CloseableHttpClient client; - //we don't rely on default headers supported by HttpClient as those cannot be replaced, plus it would get hairy - //when we create the HttpClient instance on our own as there would be two different ways to set the default headers. + private final CloseableHttpAsyncClient client; + //we don't rely on default headers supported by HttpAsyncClient as those cannot be replaced private final Header[] defaultHeaders; private final long maxRetryTimeoutMillis; private final AtomicInteger lastHostIndex = new AtomicInteger(0); @@ -89,7 +95,7 @@ public final class RestClient implements Closeable { private final ConcurrentMap blacklist = new ConcurrentHashMap<>(); private final FailureListener failureListener; - RestClient(CloseableHttpClient client, long maxRetryTimeoutMillis, Header[] defaultHeaders, + RestClient(CloseableHttpAsyncClient client, long maxRetryTimeoutMillis, Header[] defaultHeaders, HttpHost[] hosts, FailureListener failureListener) { this.client = client; this.maxRetryTimeoutMillis = maxRetryTimeoutMillis; @@ -127,8 +133,8 @@ public final class RestClient implements Closeable { * @throws ClientProtocolException in case of an http protocol error * @throws ResponseException in case elasticsearch responded with a status code that indicated an error */ - public Response performRequest(String method, String endpoint, Header... headers) throws IOException { - return performRequest(method, endpoint, Collections.emptyMap(), null, headers); + public Response performRequest(String method, String endpoint, Header... headers) throws Exception { + return performRequest(method, endpoint, Collections.emptyMap(), (HttpEntity)null, headers); } /** @@ -144,16 +150,16 @@ public final class RestClient implements Closeable { * @throws ClientProtocolException in case of an http protocol error * @throws ResponseException in case elasticsearch responded with a status code that indicated an error */ - public Response performRequest(String method, String endpoint, Map params, Header... headers) throws IOException { - return performRequest(method, endpoint, params, null, headers); + public Response performRequest(String method, String endpoint, Map params, Header... headers) throws Exception { + return performRequest(method, endpoint, params, (HttpEntity)null, headers); } /** - * Sends a request to the elasticsearch cluster that the current client points to. - * Selects a host out of the provided ones in a round-robin fashion. Failing hosts are marked dead and retried after a certain - * amount of time (minimum 1 minute, maximum 30 minutes), depending on how many times they previously failed (the more failures, - * the later they will be retried). In case of failures all of the alive nodes (or dead nodes that deserve a retry) are retried - * till one responds or none of them does, in which case an {@link IOException} will be thrown. + * Sends a request to the elasticsearch cluster that the current client points to. Blocks till the request is completed and returns + * its response of fails by throwing an exception. Selects a host out of the provided ones in a round-robin fashion. Failing hosts + * are marked dead and retried after a certain amount of time (minimum 1 minute, maximum 30 minutes), depending on how many times + * they previously failed (the more failures, the later they will be retried). In case of failures all of the alive nodes (or dead + * nodes that deserve a retry) are retried till one responds or none of them does, in which case an {@link IOException} will be thrown. * * @param method the http method * @param endpoint the path of the request (without host and port) @@ -166,73 +172,159 @@ public final class RestClient implements Closeable { * @throws ResponseException in case elasticsearch responded with a status code that indicated an error */ public Response performRequest(String method, String endpoint, Map params, - HttpEntity entity, Header... headers) throws IOException { + HttpEntity entity, Header... headers) throws Exception { + HttpAsyncResponseConsumer consumer = HttpAsyncMethods.createConsumer(); + SyncResponseListener listener = new SyncResponseListener(); + performRequest(method, endpoint, params, entity, consumer, listener, headers); + return listener.get(); + } + + /** + * Sends a request to the elasticsearch cluster that the current client points to. + * Shortcut to {@link #performRequest(String, String, Map, HttpEntity, HttpAsyncResponseConsumer, ResponseListener, Header...)} + * but without parameters, request body and async response consumer. A default response consumer, specifically an instance of + * ({@link org.apache.http.nio.protocol.BasicAsyncResponseConsumer} will be created and used. + * + * @param method the http method + * @param endpoint the path of the request (without host and port) + * @param responseListener the {@link ResponseListener} to notify when the request is completed or fails + * @param headers the optional request headers + */ + public void performRequest(String method, String endpoint, ResponseListener responseListener, Header... headers) { + performRequest(method, endpoint, Collections.emptyMap(), null, responseListener, headers); + } + + /** + * Sends a request to the elasticsearch cluster that the current client points to. + * Shortcut to {@link #performRequest(String, String, Map, HttpEntity, HttpAsyncResponseConsumer, ResponseListener, Header...)} + * but without request body and async response consumer. A default response consumer, specifically an instance of + * ({@link org.apache.http.nio.protocol.BasicAsyncResponseConsumer} will be created and used. + * + * @param method the http method + * @param endpoint the path of the request (without host and port) + * @param params the query_string parameters + * @param responseListener the {@link ResponseListener} to notify when the request is completed or fails + * @param headers the optional request headers + */ + public void performRequest(String method, String endpoint, Map params, + ResponseListener responseListener, Header... headers) { + performRequest(method, endpoint, params, null, responseListener, headers); + } + + /** + /** + * Sends a request to the elasticsearch cluster that the current client points to. + * Shortcut to {@link #performRequest(String, String, Map, HttpEntity, HttpAsyncResponseConsumer, ResponseListener, Header...)} + * but without an async response consumer, meaning that a {@link org.apache.http.nio.protocol.BasicAsyncResponseConsumer} + * will be created and used. + * + * @param method the http method + * @param endpoint the path of the request (without host and port) + * @param params the query_string parameters + * @param entity the body of the request, null if not applicable + * @param responseListener the {@link ResponseListener} to notify when the request is completed or fails + * @param headers the optional request headers + */ + public void performRequest(String method, String endpoint, Map params, + HttpEntity entity, ResponseListener responseListener, Header... headers) { + HttpAsyncResponseConsumer responseConsumer = HttpAsyncMethods.createConsumer(); + performRequest(method, endpoint, params, entity, responseConsumer, responseListener, headers); + } + + /** + * Sends a request to the elasticsearch cluster that the current client points to. The request is executed asynchronously + * and the provided {@link ResponseListener} gets notified whenever it is completed or it fails. + * Selects a host out of the provided ones in a round-robin fashion. Failing hosts are marked dead and retried after a certain + * amount of time (minimum 1 minute, maximum 30 minutes), depending on how many times they previously failed (the more failures, + * the later they will be retried). In case of failures all of the alive nodes (or dead nodes that deserve a retry) are retried + * till one responds or none of them does, in which case an {@link IOException} will be thrown. + * + * @param method the http method + * @param endpoint the path of the request (without host and port) + * @param params the query_string parameters + * @param entity the body of the request, null if not applicable + * @param responseConsumer the {@link HttpAsyncResponseConsumer} callback + * @param responseListener the {@link ResponseListener} to notify when the request is completed or fails + * @param headers the optional request headers + */ + public void performRequest(String method, String endpoint, Map params, + HttpEntity entity, HttpAsyncResponseConsumer responseConsumer, + ResponseListener responseListener, Header... headers) { URI uri = buildUri(endpoint, params); HttpRequestBase request = createHttpRequest(method, uri, entity); setHeaders(request, headers); - //we apply a soft margin so that e.g. if a request took 59 seconds and timeout is set to 60 we don't do another attempt - long retryTimeoutMillis = Math.round(this.maxRetryTimeoutMillis / (float)100 * 98); - IOException lastSeenException = null; + FailureTrackingListener failureTrackingListener = new FailureTrackingListener(responseListener); long startTime = System.nanoTime(); - for (HttpHost host : nextHost()) { - if (lastSeenException != null) { - //in case we are retrying, check whether maxRetryTimeout has been reached, in which case an exception will be thrown - long timeElapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); - long timeout = retryTimeoutMillis - timeElapsedMillis; - if (timeout <= 0) { - IOException retryTimeoutException = new IOException( - "request retries exceeded max retry timeout [" + retryTimeoutMillis + "]"); - retryTimeoutException.addSuppressed(lastSeenException); - throw retryTimeoutException; + performRequest(startTime, nextHost().iterator(), request, responseConsumer, failureTrackingListener); + } + + private void performRequest(final long startTime, final Iterator hosts, final HttpRequestBase request, + final HttpAsyncResponseConsumer responseConsumer, + final FailureTrackingListener listener) { + final HttpHost host = hosts.next(); + //we stream the request body if the entity allows for it + HttpAsyncRequestProducer requestProducer = HttpAsyncMethods.create(host, request); + client.execute(requestProducer, responseConsumer, new FutureCallback() { + @Override + public void completed(HttpResponse httpResponse) { + try { + RequestLogger.logResponse(logger, request, host, httpResponse); + int statusCode = httpResponse.getStatusLine().getStatusCode(); + Response response = new Response(request.getRequestLine(), host, httpResponse); + if (isSuccessfulResponse(request.getMethod(), statusCode)) { + onResponse(host); + listener.onSuccess(response); + } else { + ResponseException responseException = new ResponseException(response); + if (mustRetry(statusCode)) { + //mark host dead and retry against next one + onFailure(host); + retryIfPossible(responseException, hosts, request); + } else { + //mark host alive and don't retry, as the error should be a request problem + onResponse(host); + listener.onDefinitiveFailure(responseException); + } + } + } catch(Exception e) { + listener.onDefinitiveFailure(e); } - //also reset the request to make it reusable for the next attempt - request.reset(); } - CloseableHttpResponse httpResponse; - try { - httpResponse = client.execute(host, request); - } catch(IOException e) { - RequestLogger.logFailedRequest(logger, request, host, e); - onFailure(host); - lastSeenException = addSuppressedException(lastSeenException, e); - continue; - } - Response response = new Response(request.getRequestLine(), host, httpResponse); - int statusCode = response.getStatusLine().getStatusCode(); - if (statusCode < 300 || (request.getMethod().equals(HttpHead.METHOD_NAME) && statusCode == 404) ) { - RequestLogger.logResponse(logger, request, host, httpResponse); - onResponse(host); - return response; - } - RequestLogger.logResponse(logger, request, host, httpResponse); - String responseBody; - try { - if (response.getEntity() == null) { - responseBody = null; - } else { - responseBody = EntityUtils.toString(response.getEntity()); - } - } finally { - response.close(); - } - lastSeenException = addSuppressedException(lastSeenException, new ResponseException(response, responseBody)); - switch(statusCode) { - case 502: - case 503: - case 504: - //mark host dead and retry against next one + @Override + public void failed(Exception failure) { + try { + RequestLogger.logFailedRequest(logger, request, host, failure); onFailure(host); - break; - default: - //mark host alive and don't retry, as the error should be a request problem - onResponse(host); - throw lastSeenException; + retryIfPossible(failure, hosts, request); + } catch(Exception e) { + listener.onDefinitiveFailure(e); + } } - } - //we get here only when we tried all nodes and they all failed - assert lastSeenException != null; - throw lastSeenException; + + private void retryIfPossible(Exception exception, Iterator hosts, HttpRequestBase request) { + if (hosts.hasNext()) { + //in case we are retrying, check whether maxRetryTimeout has been reached + long timeElapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime); + long timeout = maxRetryTimeoutMillis - timeElapsedMillis; + if (timeout <= 0) { + IOException retryTimeoutException = new IOException( + "request retries exceeded max retry timeout [" + maxRetryTimeoutMillis + "]"); + listener.onDefinitiveFailure(retryTimeoutException); + } else { + listener.trackFailure(exception); + request.reset(); + performRequest(startTime, hosts, request, responseConsumer, listener); + } + } else { + listener.onDefinitiveFailure(exception); + } + } + + @Override + public void cancelled() { + } + }); } private void setHeaders(HttpRequest httpRequest, Header[] requestHeaders) { @@ -316,7 +408,21 @@ public final class RestClient implements Closeable { client.close(); } - private static IOException addSuppressedException(IOException suppressedException, IOException currentException) { + private static boolean isSuccessfulResponse(String method, int statusCode) { + return statusCode < 300 || (HttpHead.METHOD_NAME.equals(method) && statusCode == 404); + } + + private static boolean mustRetry(int statusCode) { + switch(statusCode) { + case 502: + case 503: + case 504: + return true; + } + return false; + } + + private static Exception addSuppressedException(Exception suppressedException, Exception currentException) { if (suppressedException != null) { currentException.addSuppressed(suppressedException); } @@ -372,6 +478,57 @@ public final class RestClient implements Closeable { } } + private static class FailureTrackingListener { + private final ResponseListener responseListener; + private volatile Exception exception; + + FailureTrackingListener(ResponseListener responseListener) { + this.responseListener = responseListener; + } + + void onSuccess(Response response) { + responseListener.onSuccess(response); + } + + void onDefinitiveFailure(Exception exception) { + trackFailure(exception); + responseListener.onFailure(this.exception); + } + + void trackFailure(Exception exception) { + this.exception = addSuppressedException(this.exception, exception); + } + } + + private static class SyncResponseListener implements ResponseListener { + final CountDownLatch latch = new CountDownLatch(1); + volatile Response response; + volatile Exception exception; + + @Override + public void onSuccess(Response response) { + this.response = response; + latch.countDown(); + + } + + @Override + public void onFailure(Exception exception) { + this.exception = exception; + latch.countDown(); + } + + Response get() throws Exception { + latch.await(); + if (response != null) { + assert exception == null; + return response; + } + assert exception != null; + throw exception; + } + } + /** * Returns a new {@link Builder} to help with {@link RestClient} creation. */ @@ -380,13 +537,17 @@ public final class RestClient implements Closeable { } /** - * Rest client builder. Helps creating a new {@link RestClient}. + * Helps creating a new {@link RestClient}. Allows to set the most common http client configuration options when internally + * creating the underlying {@link org.apache.http.nio.client.HttpAsyncClient}. Also allows to provide an externally created + * {@link org.apache.http.nio.client.HttpAsyncClient} in case additional customization is needed. */ public static final class Builder { public static final int DEFAULT_CONNECT_TIMEOUT_MILLIS = 1000; public static final int DEFAULT_SOCKET_TIMEOUT_MILLIS = 10000; public static final int DEFAULT_MAX_RETRY_TIMEOUT_MILLIS = DEFAULT_SOCKET_TIMEOUT_MILLIS; public static final int DEFAULT_CONNECTION_REQUEST_TIMEOUT_MILLIS = 500; + public static final int DEFAULT_MAX_CONN_PER_ROUTE = 10; + public static final int DEFAULT_MAX_CONN_TOTAL = 30; private static final Header[] EMPTY_HEADERS = new Header[0]; @@ -408,21 +569,7 @@ public final class RestClient implements Closeable { } /** - * Sets the maximum timeout (in milliseconds) to honour in case of multiple retries of the same request. - * {@link #DEFAULT_MAX_RETRY_TIMEOUT_MILLIS} if not specified. - * - * @throws IllegalArgumentException if maxRetryTimeoutMillis is not greater than 0 - */ - public Builder setMaxRetryTimeoutMillis(int maxRetryTimeoutMillis) { - if (maxRetryTimeoutMillis <= 0) { - throw new IllegalArgumentException("maxRetryTimeoutMillis must be greater than 0"); - } - this.maxRetryTimeout = maxRetryTimeoutMillis; - return this; - } - - /** - * Sets the default request headers, to be used sent with every request unless overridden on a per request basis + * Sets the default request headers, which will be sent along with each request */ public Builder setDefaultHeaders(Header[] defaultHeaders) { Objects.requireNonNull(defaultHeaders, "defaultHeaders must not be null"); @@ -442,6 +589,20 @@ public final class RestClient implements Closeable { return this; } + /** + * Sets the maximum timeout (in milliseconds) to honour in case of multiple retries of the same request. + * {@link #DEFAULT_MAX_RETRY_TIMEOUT_MILLIS} if not specified. + * + * @throws IllegalArgumentException if maxRetryTimeoutMillis is not greater than 0 + */ + public Builder setMaxRetryTimeoutMillis(int maxRetryTimeoutMillis) { + if (maxRetryTimeoutMillis <= 0) { + throw new IllegalArgumentException("maxRetryTimeoutMillis must be greater than 0"); + } + this.maxRetryTimeout = maxRetryTimeoutMillis; + return this; + } + /** * Sets the {@link HttpClientConfigCallback} to be used to customize http client configuration */ @@ -467,29 +628,23 @@ public final class RestClient implements Closeable { if (failureListener == null) { failureListener = new FailureListener(); } - CloseableHttpClient httpClient = createHttpClient(); + CloseableHttpAsyncClient httpClient = createHttpClient(); + httpClient.start(); return new RestClient(httpClient, maxRetryTimeout, defaultHeaders, hosts, failureListener); } - private CloseableHttpClient createHttpClient() { + private CloseableHttpAsyncClient createHttpClient() { //default timeouts are all infinite RequestConfig.Builder requestConfigBuilder = RequestConfig.custom().setConnectTimeout(DEFAULT_CONNECT_TIMEOUT_MILLIS) .setSocketTimeout(DEFAULT_SOCKET_TIMEOUT_MILLIS) .setConnectionRequestTimeout(DEFAULT_CONNECTION_REQUEST_TIMEOUT_MILLIS); - if (requestConfigCallback != null) { requestConfigCallback.customizeRequestConfig(requestConfigBuilder); } - RequestConfig requestConfig = requestConfigBuilder.build(); - - PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(); - //default settings may be too constraining - connectionManager.setDefaultMaxPerRoute(10); - connectionManager.setMaxTotal(30); - - HttpClientBuilder httpClientBuilder = HttpClientBuilder.create().setConnectionManager(connectionManager) - .setDefaultRequestConfig(requestConfig); + HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClientBuilder.create().setDefaultRequestConfig(requestConfigBuilder.build()) + //default settings for connection pooling may be too constraining + .setMaxConnPerRoute(DEFAULT_MAX_CONN_PER_ROUTE).setMaxConnTotal(DEFAULT_MAX_CONN_TOTAL); if (httpClientConfigCallback != null) { httpClientConfigCallback.customizeHttpClient(httpClientBuilder); } @@ -517,12 +672,12 @@ public final class RestClient implements Closeable { */ public interface HttpClientConfigCallback { /** - * Allows to customize the {@link CloseableHttpClient} being created and used by the {@link RestClient}. - * It is common to customzie the default {@link org.apache.http.client.CredentialsProvider} through this method, - * without losing any other useful default value that the {@link RestClient.Builder} internally sets. - * Also useful to setup ssl through {@link SSLSocketFactoryHttpConfigCallback}. + * Allows to customize the {@link CloseableHttpAsyncClient} being created and used by the {@link RestClient}. + * Commonly used to customize the default {@link org.apache.http.client.CredentialsProvider} for authentication + * or the {@link SchemeIOSessionStrategy} for communication through ssl without losing any other useful default + * value that the {@link RestClient.Builder} internally sets, like connection pooling. */ - void customizeHttpClient(HttpClientBuilder httpClientBuilder); + void customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder); } /** @@ -533,7 +688,7 @@ public final class RestClient implements Closeable { /** * Notifies that the host provided as argument has just failed */ - public void onFailure(HttpHost host) throws IOException { + public void onFailure(HttpHost host) { } } diff --git a/client/rest/src/main/java/org/elasticsearch/client/SSLSocketFactoryHttpConfigCallback.java b/client/rest/src/main/java/org/elasticsearch/client/SSLSocketFactoryHttpConfigCallback.java deleted file mode 100644 index 3f18f9938c1..00000000000 --- a/client/rest/src/main/java/org/elasticsearch/client/SSLSocketFactoryHttpConfigCallback.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Licensed to Elasticsearch under one or more contributor - * license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright - * ownership. Elasticsearch licenses this file to you under - * the Apache License, Version 2.0 (the "License"); you may - * not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.elasticsearch.client; - -import org.apache.http.config.Registry; -import org.apache.http.config.RegistryBuilder; -import org.apache.http.conn.socket.ConnectionSocketFactory; -import org.apache.http.conn.socket.PlainConnectionSocketFactory; -import org.apache.http.conn.ssl.SSLConnectionSocketFactory; -import org.apache.http.impl.client.HttpClientBuilder; -import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; - -/** - * Helps configuring the http client when needing to communicate over ssl. It effectively replaces the connection manager - * with one that has ssl properly configured thanks to the provided {@link SSLConnectionSocketFactory}. - */ -public class SSLSocketFactoryHttpConfigCallback implements RestClient.HttpClientConfigCallback { - - private final SSLConnectionSocketFactory sslSocketFactory; - - public SSLSocketFactoryHttpConfigCallback(SSLConnectionSocketFactory sslSocketFactory) { - this.sslSocketFactory = sslSocketFactory; - } - - @Override - public void customizeHttpClient(HttpClientBuilder httpClientBuilder) { - Registry socketFactoryRegistry = RegistryBuilder.create() - .register("http", PlainConnectionSocketFactory.getSocketFactory()) - .register("https", sslSocketFactory).build(); - PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(socketFactoryRegistry); - //default settings may be too constraining - connectionManager.setDefaultMaxPerRoute(10); - connectionManager.setMaxTotal(30); - httpClientBuilder.setConnectionManager(connectionManager); - } -} diff --git a/client/rest/src/test/java/org/elasticsearch/client/RestClientBuilderTests.java b/client/rest/src/test/java/org/elasticsearch/client/RestClientBuilderTests.java index f032933db24..dc1c03b06a3 100644 --- a/client/rest/src/test/java/org/elasticsearch/client/RestClientBuilderTests.java +++ b/client/rest/src/test/java/org/elasticsearch/client/RestClientBuilderTests.java @@ -23,7 +23,7 @@ import com.carrotsearch.randomizedtesting.generators.RandomInts; import org.apache.http.Header; import org.apache.http.HttpHost; import org.apache.http.client.config.RequestConfig; -import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; import org.apache.http.message.BasicHeader; import java.io.IOException; @@ -108,7 +108,7 @@ public class RestClientBuilderTests extends RestClientTestCase { if (getRandom().nextBoolean()) { builder.setHttpClientConfigCallback(new RestClient.HttpClientConfigCallback() { @Override - public void customizeHttpClient(HttpClientBuilder httpClientBuilder) { + public void customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) { } }); } diff --git a/client/rest/src/test/java/org/elasticsearch/client/RestClientIntegTests.java b/client/rest/src/test/java/org/elasticsearch/client/RestClientIntegTests.java index 4a14c174353..b1f9b66557b 100644 --- a/client/rest/src/test/java/org/elasticsearch/client/RestClientIntegTests.java +++ b/client/rest/src/test/java/org/elasticsearch/client/RestClientIntegTests.java @@ -27,6 +27,7 @@ import com.sun.net.httpserver.HttpHandler; import com.sun.net.httpserver.HttpServer; import org.apache.http.Consts; import org.apache.http.Header; +import org.apache.http.HttpEntity; import org.apache.http.HttpHost; import org.apache.http.entity.StringEntity; import org.apache.http.message.BasicHeader; @@ -143,7 +144,7 @@ public class RestClientIntegTests extends RestClientTestCase { public void testHeaders() throws Exception { for (String method : getHttpMethods()) { Set standardHeaders = new HashSet<>( - Arrays.asList("Accept-encoding", "Connection", "Host", "User-agent", "Date")); + Arrays.asList("Connection", "Host", "User-agent", "Date")); if (method.equals("HEAD") == false) { standardHeaders.add("Content-length"); } @@ -162,9 +163,9 @@ public class RestClientIntegTests extends RestClientTestCase { int statusCode = randomStatusCode(getRandom()); Response esResponse; - try (Response response = restClient.performRequest(method, "/" + statusCode, - Collections.emptyMap(), null, headers)) { - esResponse = response; + try { + esResponse = restClient.performRequest(method, "/" + statusCode, Collections.emptyMap(), + (HttpEntity)null, headers); } catch(ResponseException e) { esResponse = e.getResponse(); } @@ -204,18 +205,14 @@ public class RestClientIntegTests extends RestClientTestCase { private void bodyTest(String method) throws Exception { String requestBody = "{ \"field\": \"value\" }"; StringEntity entity = new StringEntity(requestBody); - Response esResponse; - String responseBody; int statusCode = randomStatusCode(getRandom()); - try (Response response = restClient.performRequest(method, "/" + statusCode, - Collections.emptyMap(), entity)) { - responseBody = EntityUtils.toString(response.getEntity()); - esResponse = response; + Response esResponse; + try { + esResponse = restClient.performRequest(method, "/" + statusCode, Collections.emptyMap(), entity); } catch(ResponseException e) { - responseBody = e.getResponseBody(); esResponse = e.getResponse(); } assertEquals(statusCode, esResponse.getStatusLine().getStatusCode()); - assertEquals(requestBody, responseBody); + assertEquals(requestBody, EntityUtils.toString(esResponse.getEntity())); } } diff --git a/client/rest/src/test/java/org/elasticsearch/client/RestClientMultipleHostsTests.java b/client/rest/src/test/java/org/elasticsearch/client/RestClientMultipleHostsTests.java index 1a469f2e2ca..957802634bc 100644 --- a/client/rest/src/test/java/org/elasticsearch/client/RestClientMultipleHostsTests.java +++ b/client/rest/src/test/java/org/elasticsearch/client/RestClientMultipleHostsTests.java @@ -22,14 +22,17 @@ package org.elasticsearch.client; import com.carrotsearch.randomizedtesting.generators.RandomInts; import org.apache.http.Header; import org.apache.http.HttpHost; -import org.apache.http.HttpRequest; +import org.apache.http.HttpResponse; import org.apache.http.ProtocolVersion; import org.apache.http.StatusLine; -import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.concurrent.FutureCallback; import org.apache.http.conn.ConnectTimeoutException; -import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; +import org.apache.http.message.BasicHttpResponse; import org.apache.http.message.BasicStatusLine; +import org.apache.http.nio.protocol.HttpAsyncRequestProducer; +import org.apache.http.nio.protocol.HttpAsyncResponseConsumer; import org.junit.Before; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -39,6 +42,7 @@ import java.net.SocketTimeoutException; import java.util.Collections; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.Future; import static org.elasticsearch.client.RestClientTestUtil.randomErrorNoRetryStatusCode; import static org.elasticsearch.client.RestClientTestUtil.randomErrorRetryStatusCode; @@ -65,27 +69,33 @@ public class RestClientMultipleHostsTests extends RestClientTestCase { private TrackingFailureListener failureListener; @Before + @SuppressWarnings("unchecked") public void createRestClient() throws IOException { - CloseableHttpClient httpClient = mock(CloseableHttpClient.class); - when(httpClient.execute(any(HttpHost.class), any(HttpRequest.class))).thenAnswer(new Answer() { + CloseableHttpAsyncClient httpClient = mock(CloseableHttpAsyncClient.class); + when(httpClient.execute(any(HttpAsyncRequestProducer.class), any(HttpAsyncResponseConsumer.class), + any(FutureCallback.class))).thenAnswer(new Answer>() { @Override - public CloseableHttpResponse answer(InvocationOnMock invocationOnMock) throws Throwable { - HttpHost httpHost = (HttpHost) invocationOnMock.getArguments()[0]; - HttpUriRequest request = (HttpUriRequest) invocationOnMock.getArguments()[1]; + public Future answer(InvocationOnMock invocationOnMock) throws Throwable { + HttpAsyncRequestProducer requestProducer = (HttpAsyncRequestProducer) invocationOnMock.getArguments()[0]; + HttpUriRequest request = (HttpUriRequest)requestProducer.generateRequest(); + HttpHost httpHost = requestProducer.getTarget(); + @SuppressWarnings("unchecked") + FutureCallback futureCallback = (FutureCallback) invocationOnMock.getArguments()[2]; //return the desired status code or exception depending on the path if (request.getURI().getPath().equals("/soe")) { - throw new SocketTimeoutException(httpHost.toString()); + futureCallback.failed(new SocketTimeoutException(httpHost.toString())); } else if (request.getURI().getPath().equals("/coe")) { - throw new ConnectTimeoutException(httpHost.toString()); + futureCallback.failed(new ConnectTimeoutException(httpHost.toString())); } else if (request.getURI().getPath().equals("/ioe")) { - throw new IOException(httpHost.toString()); + futureCallback.failed(new IOException(httpHost.toString())); + } else { + int statusCode = Integer.parseInt(request.getURI().getPath().substring(1)); + StatusLine statusLine = new BasicStatusLine(new ProtocolVersion("http", 1, 1), statusCode, ""); + futureCallback.completed(new BasicHttpResponse(statusLine)); } - int statusCode = Integer.parseInt(request.getURI().getPath().substring(1)); - StatusLine statusLine = new BasicStatusLine(new ProtocolVersion("http", 1, 1), statusCode, ""); - return new CloseableBasicHttpResponse(statusLine); + return null; } }); - int numHosts = RandomInts.randomIntBetween(getRandom(), 2, 5); httpHosts = new HttpHost[numHosts]; for (int i = 0; i < numHosts; i++) { @@ -102,10 +112,9 @@ public class RestClientMultipleHostsTests extends RestClientTestCase { Collections.addAll(hostsSet, httpHosts); for (int j = 0; j < httpHosts.length; j++) { int statusCode = randomOkStatusCode(getRandom()); - try (Response response = restClient.performRequest(randomHttpMethod(getRandom()), "/" + statusCode)) { - assertThat(response.getStatusLine().getStatusCode(), equalTo(statusCode)); - assertTrue("host not found: " + response.getHost(), hostsSet.remove(response.getHost())); - } + Response response = restClient.performRequest(randomHttpMethod(getRandom()), "/" + statusCode); + assertEquals(statusCode, response.getStatusLine().getStatusCode()); + assertTrue("host not found: " + response.getHost(), hostsSet.remove(response.getHost())); } assertEquals("every host should have been used but some weren't: " + hostsSet, 0, hostsSet.size()); } @@ -120,11 +129,12 @@ public class RestClientMultipleHostsTests extends RestClientTestCase { for (int j = 0; j < httpHosts.length; j++) { String method = randomHttpMethod(getRandom()); int statusCode = randomErrorNoRetryStatusCode(getRandom()); - try (Response response = restClient.performRequest(method, "/" + statusCode)) { + try { + Response response = restClient.performRequest(method, "/" + statusCode); if (method.equals("HEAD") && statusCode == 404) { //no exception gets thrown although we got a 404 - assertThat(response.getStatusLine().getStatusCode(), equalTo(404)); - assertThat(response.getStatusLine().getStatusCode(), equalTo(statusCode)); + assertEquals(404, response.getStatusLine().getStatusCode()); + assertEquals(statusCode, response.getStatusLine().getStatusCode()); assertTrue("host not found: " + response.getHost(), hostsSet.remove(response.getHost())); } else { fail("request should have failed"); @@ -134,7 +144,7 @@ public class RestClientMultipleHostsTests extends RestClientTestCase { throw e; } Response response = e.getResponse(); - assertThat(response.getStatusLine().getStatusCode(), equalTo(statusCode)); + assertEquals(statusCode, response.getStatusLine().getStatusCode()); assertTrue("host not found: " + response.getHost(), hostsSet.remove(response.getHost())); assertEquals(0, e.getSuppressed().length); } @@ -156,7 +166,7 @@ public class RestClientMultipleHostsTests extends RestClientTestCase { failureListener.assertCalled(httpHosts); do { Response response = e.getResponse(); - assertThat(response.getStatusLine().getStatusCode(), equalTo(Integer.parseInt(retryEndpoint.substring(1)))); + assertEquals(Integer.parseInt(retryEndpoint.substring(1)), response.getStatusLine().getStatusCode()); assertTrue("host [" + response.getHost() + "] not found, most likely used multiple times", hostsSet.remove(response.getHost())); if (e.getSuppressed().length > 0) { @@ -223,8 +233,8 @@ public class RestClientMultipleHostsTests extends RestClientTestCase { for (int y = 0; y < iters; y++) { int statusCode = randomErrorNoRetryStatusCode(getRandom()); Response response; - try (Response esResponse = restClient.performRequest(randomHttpMethod(getRandom()), "/" + statusCode)) { - response = esResponse; + try { + response = restClient.performRequest(randomHttpMethod(getRandom()), "/" + statusCode); } catch(ResponseException e) { response = e.getResponse(); diff --git a/client/rest/src/test/java/org/elasticsearch/client/RestClientSingleHostTests.java b/client/rest/src/test/java/org/elasticsearch/client/RestClientSingleHostTests.java index e7c9de934b6..403b86753c7 100644 --- a/client/rest/src/test/java/org/elasticsearch/client/RestClientSingleHostTests.java +++ b/client/rest/src/test/java/org/elasticsearch/client/RestClientSingleHostTests.java @@ -26,9 +26,9 @@ import org.apache.http.HttpEntity; import org.apache.http.HttpEntityEnclosingRequest; import org.apache.http.HttpHost; import org.apache.http.HttpRequest; +import org.apache.http.HttpResponse; import org.apache.http.ProtocolVersion; import org.apache.http.StatusLine; -import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpHead; import org.apache.http.client.methods.HttpOptions; import org.apache.http.client.methods.HttpPatch; @@ -37,11 +37,15 @@ import org.apache.http.client.methods.HttpPut; import org.apache.http.client.methods.HttpTrace; import org.apache.http.client.methods.HttpUriRequest; import org.apache.http.client.utils.URIBuilder; +import org.apache.http.concurrent.FutureCallback; import org.apache.http.conn.ConnectTimeoutException; import org.apache.http.entity.StringEntity; -import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; import org.apache.http.message.BasicHeader; +import org.apache.http.message.BasicHttpResponse; import org.apache.http.message.BasicStatusLine; +import org.apache.http.nio.protocol.HttpAsyncRequestProducer; +import org.apache.http.nio.protocol.HttpAsyncResponseConsumer; import org.apache.http.util.EntityUtils; import org.junit.Before; import org.mockito.ArgumentCaptor; @@ -51,11 +55,11 @@ import org.mockito.stubbing.Answer; import java.io.IOException; import java.net.SocketTimeoutException; import java.net.URI; -import java.net.URISyntaxException; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.Future; import static org.elasticsearch.client.RestClientTestUtil.getAllErrorStatusCodes; import static org.elasticsearch.client.RestClientTestUtil.getHttpMethods; @@ -86,39 +90,49 @@ public class RestClientSingleHostTests extends RestClientTestCase { private RestClient restClient; private Header[] defaultHeaders; private HttpHost httpHost; - private CloseableHttpClient httpClient; + private CloseableHttpAsyncClient httpClient; private TrackingFailureListener failureListener; @Before + @SuppressWarnings("unchecked") public void createRestClient() throws IOException { - httpClient = mock(CloseableHttpClient.class); - when(httpClient.execute(any(HttpHost.class), any(HttpRequest.class))).thenAnswer(new Answer() { - @Override - public CloseableHttpResponse answer(InvocationOnMock invocationOnMock) throws Throwable { - HttpUriRequest request = (HttpUriRequest) invocationOnMock.getArguments()[1]; - //return the desired status code or exception depending on the path - if (request.getURI().getPath().equals("/soe")) { - throw new SocketTimeoutException(); - } else if (request.getURI().getPath().equals("/coe")) { - throw new ConnectTimeoutException(); - } - int statusCode = Integer.parseInt(request.getURI().getPath().substring(1)); - StatusLine statusLine = new BasicStatusLine(new ProtocolVersion("http", 1, 1), statusCode, ""); + httpClient = mock(CloseableHttpAsyncClient.class); + when(httpClient.execute(any(HttpAsyncRequestProducer.class), any(HttpAsyncResponseConsumer.class), + any(FutureCallback.class))).thenAnswer(new Answer>() { + @Override + public Future answer(InvocationOnMock invocationOnMock) throws Throwable { + HttpAsyncRequestProducer requestProducer = (HttpAsyncRequestProducer) invocationOnMock.getArguments()[0]; + @SuppressWarnings("unchecked") + FutureCallback futureCallback = (FutureCallback) invocationOnMock.getArguments()[2]; + HttpUriRequest request = (HttpUriRequest)requestProducer.generateRequest(); + //return the desired status code or exception depending on the path + if (request.getURI().getPath().equals("/soe")) { + futureCallback.failed(new SocketTimeoutException()); + } else if (request.getURI().getPath().equals("/coe")) { + futureCallback.failed(new ConnectTimeoutException()); + } else { + int statusCode = Integer.parseInt(request.getURI().getPath().substring(1)); + StatusLine statusLine = new BasicStatusLine(new ProtocolVersion("http", 1, 1), statusCode, ""); - CloseableHttpResponse httpResponse = new CloseableBasicHttpResponse(statusLine); - //return the same body that was sent - if (request instanceof HttpEntityEnclosingRequest) { - HttpEntity entity = ((HttpEntityEnclosingRequest) request).getEntity(); - if (entity != null) { - assertTrue("the entity is not repeatable, cannot set it to the response directly", entity.isRepeatable()); - httpResponse.setEntity(entity); + HttpResponse httpResponse = new BasicHttpResponse(statusLine); + //return the same body that was sent + if (request instanceof HttpEntityEnclosingRequest) { + HttpEntity entity = ((HttpEntityEnclosingRequest) request).getEntity(); + if (entity != null) { + assertTrue("the entity is not repeatable, cannot set it to the response directly", + entity.isRepeatable()); + httpResponse.setEntity(entity); + } + } + //return the same headers that were sent + httpResponse.setHeaders(request.getAllHeaders()); + futureCallback.completed(httpResponse); + } + return null; } - } - //return the same headers that were sent - httpResponse.setHeaders(request.getAllHeaders()); - return httpResponse; - } - }); + }); + + int numHeaders = RandomInts.randomIntBetween(getRandom(), 0, 3); defaultHeaders = new Header[numHeaders]; for (int i = 0; i < numHeaders; i++) { @@ -134,13 +148,15 @@ public class RestClientSingleHostTests extends RestClientTestCase { /** * Verifies the content of the {@link HttpRequest} that's internally created and passed through to the http client */ + @SuppressWarnings("unchecked") public void testInternalHttpRequest() throws Exception { - ArgumentCaptor requestArgumentCaptor = ArgumentCaptor.forClass(HttpUriRequest.class); + ArgumentCaptor requestArgumentCaptor = ArgumentCaptor.forClass(HttpAsyncRequestProducer.class); int times = 0; for (String httpMethod : getHttpMethods()) { HttpUriRequest expectedRequest = performRandomRequest(httpMethod); - verify(httpClient, times(++times)).execute(any(HttpHost.class), requestArgumentCaptor.capture()); - HttpUriRequest actualRequest = requestArgumentCaptor.getValue(); + verify(httpClient, times(++times)).execute(requestArgumentCaptor.capture(), + any(HttpAsyncResponseConsumer.class), any(FutureCallback.class)); + HttpUriRequest actualRequest = (HttpUriRequest)requestArgumentCaptor.getValue().generateRequest(); assertEquals(expectedRequest.getURI(), actualRequest.getURI()); assertEquals(expectedRequest.getClass(), actualRequest.getClass()); assertArrayEquals(expectedRequest.getAllHeaders(), actualRequest.getAllHeaders()); @@ -201,7 +217,8 @@ public class RestClientSingleHostTests extends RestClientTestCase { for (String method : getHttpMethods()) { //error status codes should cause an exception to be thrown for (int errorStatusCode : getAllErrorStatusCodes()) { - try (Response response = performRequest(method, "/" + errorStatusCode)) { + try { + Response response = performRequest(method, "/" + errorStatusCode); if (method.equals("HEAD") && errorStatusCode == 404) { //no exception gets thrown although we got a 404 assertThat(response.getStatusLine().getStatusCode(), equalTo(errorStatusCode)); @@ -223,7 +240,7 @@ public class RestClientSingleHostTests extends RestClientTestCase { } } - public void testIOExceptions() throws IOException { + public void testIOExceptions() throws Exception { for (String method : getHttpMethods()) { //IOExceptions should be let bubble up try { @@ -252,11 +269,9 @@ public class RestClientSingleHostTests extends RestClientTestCase { StringEntity entity = new StringEntity(body); for (String method : Arrays.asList("DELETE", "GET", "PATCH", "POST", "PUT")) { for (int okStatusCode : getOkStatusCodes()) { - try (Response response = restClient.performRequest(method, "/" + okStatusCode, - Collections.emptyMap(), entity)) { - assertThat(response.getStatusLine().getStatusCode(), equalTo(okStatusCode)); - assertThat(EntityUtils.toString(response.getEntity()), equalTo(body)); - } + Response response = restClient.performRequest(method, "/" + okStatusCode, Collections.emptyMap(), entity); + assertThat(response.getStatusLine().getStatusCode(), equalTo(okStatusCode)); + assertThat(EntityUtils.toString(response.getEntity()), equalTo(body)); } for (int errorStatusCode : getAllErrorStatusCodes()) { try { @@ -334,9 +349,8 @@ public class RestClientSingleHostTests extends RestClientTestCase { int statusCode = randomStatusCode(getRandom()); Response esResponse; - try (Response response = restClient.performRequest(method, "/" + statusCode, - Collections.emptyMap(), null, headers)) { - esResponse = response; + try { + esResponse = restClient.performRequest(method, "/" + statusCode, headers); } catch(ResponseException e) { esResponse = e.getResponse(); } @@ -349,7 +363,7 @@ public class RestClientSingleHostTests extends RestClientTestCase { } } - private HttpUriRequest performRandomRequest(String method) throws IOException, URISyntaxException { + private HttpUriRequest performRandomRequest(String method) throws Exception { String uriAsString = "/" + randomStatusCode(getRandom()); URIBuilder uriBuilder = new URIBuilder(uriAsString); Map params = Collections.emptyMap(); @@ -434,14 +448,14 @@ public class RestClientSingleHostTests extends RestClientTestCase { return request; } - private Response performRequest(String method, String endpoint, Header... headers) throws IOException { + private Response performRequest(String method, String endpoint, Header... headers) throws Exception { switch(randomIntBetween(0, 2)) { case 0: return restClient.performRequest(method, endpoint, headers); case 1: return restClient.performRequest(method, endpoint, Collections.emptyMap(), headers); case 2: - return restClient.performRequest(method, endpoint, Collections.emptyMap(), null, headers); + return restClient.performRequest(method, endpoint, Collections.emptyMap(), (HttpEntity)null, headers); default: throw new UnsupportedOperationException(); } diff --git a/client/rest/src/test/java/org/elasticsearch/client/TrackingFailureListener.java b/client/rest/src/test/java/org/elasticsearch/client/TrackingFailureListener.java index 35842823923..92033b72cef 100644 --- a/client/rest/src/test/java/org/elasticsearch/client/TrackingFailureListener.java +++ b/client/rest/src/test/java/org/elasticsearch/client/TrackingFailureListener.java @@ -21,7 +21,6 @@ package org.elasticsearch.client; import org.apache.http.HttpHost; -import java.io.IOException; import java.util.HashSet; import java.util.Set; @@ -33,10 +32,10 @@ import static org.junit.Assert.assertThat; * {@link org.elasticsearch.client.RestClient.FailureListener} impl that allows to track when it gets called */ class TrackingFailureListener extends RestClient.FailureListener { - private Set hosts = new HashSet<>(); + private volatile Set hosts = new HashSet<>(); @Override - public void onFailure(HttpHost host) throws IOException { + public void onFailure(HttpHost host) { hosts.add(host); } diff --git a/client/sniffer/src/main/java/org/elasticsearch/client/sniff/HostsSniffer.java b/client/sniffer/src/main/java/org/elasticsearch/client/sniff/HostsSniffer.java index bfe21f5e7d1..4aae68caef0 100644 --- a/client/sniffer/src/main/java/org/elasticsearch/client/sniff/HostsSniffer.java +++ b/client/sniffer/src/main/java/org/elasticsearch/client/sniff/HostsSniffer.java @@ -61,10 +61,9 @@ public class HostsSniffer { /** * Calls the elasticsearch nodes info api, parses the response and returns all the found http hosts */ - public List sniffHosts() throws IOException { - try (Response response = restClient.performRequest("get", "/_nodes/http", sniffRequestParams)) { - return readHosts(response.getEntity()); - } + public List sniffHosts() throws Exception { + Response response = restClient.performRequest("get", "/_nodes/http", sniffRequestParams); + return readHosts(response.getEntity()); } private List readHosts(HttpEntity entity) throws IOException { diff --git a/client/sniffer/src/main/java/org/elasticsearch/client/sniff/SniffOnFailureListener.java b/client/sniffer/src/main/java/org/elasticsearch/client/sniff/SniffOnFailureListener.java index 76350057141..cbc77351de9 100644 --- a/client/sniffer/src/main/java/org/elasticsearch/client/sniff/SniffOnFailureListener.java +++ b/client/sniffer/src/main/java/org/elasticsearch/client/sniff/SniffOnFailureListener.java @@ -22,7 +22,6 @@ package org.elasticsearch.client.sniff; import org.apache.http.HttpHost; import org.elasticsearch.client.RestClient; -import java.io.IOException; import java.util.Objects; import java.util.concurrent.atomic.AtomicBoolean; @@ -55,7 +54,7 @@ public class SniffOnFailureListener extends RestClient.FailureListener { } @Override - public void onFailure(HttpHost host) throws IOException { + public void onFailure(HttpHost host) { if (sniffer == null) { throw new IllegalStateException("sniffer was not set, unable to sniff on failure"); } diff --git a/client/sniffer/src/test/java/org/elasticsearch/client/sniff/HostsSnifferTests.java b/client/sniffer/src/test/java/org/elasticsearch/client/sniff/HostsSnifferTests.java index 6e0c3a728d5..5a9fd4033d1 100644 --- a/client/sniffer/src/test/java/org/elasticsearch/client/sniff/HostsSnifferTests.java +++ b/client/sniffer/src/test/java/org/elasticsearch/client/sniff/HostsSnifferTests.java @@ -43,7 +43,6 @@ import java.io.OutputStream; import java.io.StringWriter; import java.net.InetAddress; import java.net.InetSocketAddress; -import java.net.URISyntaxException; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -86,7 +85,7 @@ public class HostsSnifferTests extends RestClientTestCase { httpServer.stop(0); } - public void testSniffNodes() throws IOException, URISyntaxException { + public void testSniffNodes() throws Exception { HttpHost httpHost = new HttpHost(httpServer.getAddress().getHostString(), httpServer.getAddress().getPort()); try (RestClient restClient = RestClient.builder(httpHost).build()) { HostsSniffer.Builder builder = HostsSniffer.builder(restClient).setSniffRequestTimeoutMillis(sniffRequestTimeout); diff --git a/client/sniffer/src/test/java/org/elasticsearch/client/sniff/SniffOnFailureListenerTests.java b/client/sniffer/src/test/java/org/elasticsearch/client/sniff/SniffOnFailureListenerTests.java index 6a71d72f60e..bbb1de35663 100644 --- a/client/sniffer/src/test/java/org/elasticsearch/client/sniff/SniffOnFailureListenerTests.java +++ b/client/sniffer/src/test/java/org/elasticsearch/client/sniff/SniffOnFailureListenerTests.java @@ -45,16 +45,17 @@ public class SniffOnFailureListenerTests extends RestClientTestCase { assertEquals("sniffer must not be null", e.getMessage()); } - RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build(); - try (Sniffer sniffer = Sniffer.builder(restClient, new MockHostsSniffer()).build()) { - listener.setSniffer(sniffer); - try { + try (RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build()) { + try (Sniffer sniffer = Sniffer.builder(restClient, new MockHostsSniffer()).build()) { listener.setSniffer(sniffer); - fail("should have failed"); - } catch(IllegalStateException e) { - assertEquals("sniffer can only be set once", e.getMessage()); + try { + listener.setSniffer(sniffer); + fail("should have failed"); + } catch(IllegalStateException e) { + assertEquals("sniffer can only be set once", e.getMessage()); + } + listener.onFailure(new HttpHost("localhost", 9200)); } - listener.onFailure(new HttpHost("localhost", 9200)); } } } diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteScrollableHitSource.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteScrollableHitSource.java index 62dbd59f80a..5a08ab6a2e7 100644 --- a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteScrollableHitSource.java +++ b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteScrollableHitSource.java @@ -214,10 +214,9 @@ public class RemoteScrollableHitSource extends ScrollableHitSource { threadPool.generic().execute(new AbstractRunnable() { @Override protected void doRun() throws Exception { - try (org.elasticsearch.client.Response response = restClient.performRequest(method, uri, params, entity)) { - InputStream markSupportedInputStream = new BufferedInputStream(response.getEntity().getContent()); - listener.onResponse(markSupportedInputStream); - } + org.elasticsearch.client.Response response = restClient.performRequest(method, uri, params, entity); + InputStream markSupportedInputStream = new BufferedInputStream(response.getEntity().getContent()); + listener.onResponse(markSupportedInputStream); } @Override diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ContextAndHeaderTransportIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ContextAndHeaderTransportIT.java index caaa328b1ab..f39abb445d8 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ContextAndHeaderTransportIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ContextAndHeaderTransportIT.java @@ -220,18 +220,15 @@ public class ContextAndHeaderTransportIT extends HttpSmokeTestCase { public void testThatRelevantHttpHeadersBecomeRequestHeaders() throws Exception { final String IRRELEVANT_HEADER = "SomeIrrelevantHeader"; - - try (Response response = getRestClient().performRequest( - "GET", "/" + queryIndex + "/_search", - new BasicHeader(CUSTOM_HEADER, randomHeaderValue), new BasicHeader(IRRELEVANT_HEADER, randomHeaderValue))) { - assertThat(response.getStatusLine().getStatusCode(), equalTo(200)); - List searchRequests = getRequests(SearchRequest.class); - assertThat(searchRequests, hasSize(greaterThan(0))); - for (RequestAndHeaders requestAndHeaders : searchRequests) { - assertThat(requestAndHeaders.headers.containsKey(CUSTOM_HEADER), is(true)); - // was not specified, thus is not included - assertThat(requestAndHeaders.headers.containsKey(IRRELEVANT_HEADER), is(false)); - } + Response response = getRestClient().performRequest("GET", "/" + queryIndex + "/_search", + new BasicHeader(CUSTOM_HEADER, randomHeaderValue), new BasicHeader(IRRELEVANT_HEADER, randomHeaderValue)); + assertThat(response.getStatusLine().getStatusCode(), equalTo(200)); + List searchRequests = getRequests(SearchRequest.class); + assertThat(searchRequests, hasSize(greaterThan(0))); + for (RequestAndHeaders requestAndHeaders : searchRequests) { + assertThat(requestAndHeaders.headers.containsKey(CUSTOM_HEADER), is(true)); + // was not specified, thus is not included + assertThat(requestAndHeaders.headers.containsKey(IRRELEVANT_HEADER), is(false)); } } diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsNotSetIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsNotSetIT.java index 7cc84354f6f..576d5c0db96 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsNotSetIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsNotSetIT.java @@ -28,22 +28,19 @@ import static org.hamcrest.Matchers.nullValue; public class CorsNotSetIT extends HttpSmokeTestCase { - public void testCorsSettingDefaultBehaviourDoesNotReturnAnything() throws Exception { String corsValue = "http://localhost:9200"; - try (Response response = getRestClient().performRequest("GET", "/", - new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue))) { - assertThat(response.getStatusLine().getStatusCode(), is(200)); - assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); - assertThat(response.getHeader("Access-Control-Allow-Credentials"), nullValue()); - } + Response response = getRestClient().performRequest("GET", "/", + new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue)); + assertThat(response.getStatusLine().getStatusCode(), is(200)); + assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); + assertThat(response.getHeader("Access-Control-Allow-Credentials"), nullValue()); } public void testThatOmittingCorsHeaderDoesNotReturnAnything() throws Exception { - try (Response response = getRestClient().performRequest("GET", "/")) { - assertThat(response.getStatusLine().getStatusCode(), is(200)); - assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); - assertThat(response.getHeader("Access-Control-Allow-Credentials"), nullValue()); - } + Response response = getRestClient().performRequest("GET", "/"); + assertThat(response.getStatusLine().getStatusCode(), is(200)); + assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); + assertThat(response.getHeader("Access-Control-Allow-Credentials"), nullValue()); } } diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsRegexIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsRegexIT.java index a7d3f4156df..5bcef4828c9 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsRegexIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/CorsRegexIT.java @@ -26,6 +26,7 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.test.ESIntegTestCase; import org.elasticsearch.test.ESIntegTestCase.ClusterScope; import org.elasticsearch.test.ESIntegTestCase.Scope; +import org.jboss.netty.handler.codec.http.HttpHeaders; import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_CREDENTIALS; import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_METHODS; @@ -59,16 +60,15 @@ public class CorsRegexIT extends HttpSmokeTestCase { public void testThatRegularExpressionWorksOnMatch() throws Exception { String corsValue = "http://localhost:9200"; - try (Response response = getRestClient().performRequest("GET", "/", - new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue))) { - assertResponseWithOriginheader(response, corsValue); - } + Response response = getRestClient().performRequest("GET", "/", + new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue)); + assertResponseWithOriginheader(response, corsValue); + corsValue = "https://localhost:9200"; - try (Response response = getRestClient().performRequest("GET", "/", - new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue));) { - assertResponseWithOriginheader(response, corsValue); - assertThat(response.getHeader("Access-Control-Allow-Credentials"), is("true")); - } + response = getRestClient().performRequest("GET", "/", + new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue)); + assertResponseWithOriginheader(response, corsValue); + assertThat(response.getHeader("Access-Control-Allow-Credentials"), is("true")); } public void testThatRegularExpressionReturnsForbiddenOnNonMatch() throws Exception { @@ -85,27 +85,24 @@ public class CorsRegexIT extends HttpSmokeTestCase { } public void testThatSendingNoOriginHeaderReturnsNoAccessControlHeader() throws Exception { - try (Response response = getRestClient().performRequest("GET", "/", new BasicHeader("User-Agent", "Mozilla Bar"))) { - assertThat(response.getStatusLine().getStatusCode(), is(200)); - assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); - } + Response response = getRestClient().performRequest("GET", "/", new BasicHeader("User-Agent", "Mozilla Bar")); + assertThat(response.getStatusLine().getStatusCode(), is(200)); + assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); } public void testThatRegularExpressionIsNotAppliedWithoutCorrectBrowserOnMatch() throws Exception { - try (Response response = getRestClient().performRequest("GET", "/")) { - assertThat(response.getStatusLine().getStatusCode(), is(200)); - assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); - } + Response response = getRestClient().performRequest("GET", "/"); + assertThat(response.getStatusLine().getStatusCode(), is(200)); + assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue()); } public void testThatPreFlightRequestWorksOnMatch() throws Exception { String corsValue = "http://localhost:9200"; - try (Response response = getRestClient().performRequest("OPTIONS", "/", + Response response = getRestClient().performRequest("OPTIONS", "/", new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue), - new BasicHeader("Access-Control-Request-Method", "GET"));) { - assertResponseWithOriginheader(response, corsValue); - assertNotNull(response.getHeader("Access-Control-Allow-Methods")); - } + new BasicHeader(HttpHeaders.Names.ACCESS_CONTROL_REQUEST_METHOD, "GET")); + assertResponseWithOriginheader(response, corsValue); + assertNotNull(response.getHeader("Access-Control-Allow-Methods")); } public void testThatPreFlightRequestReturnsNullOnNonMatch() throws Exception { diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DeprecationHttpIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DeprecationHttpIT.java index f3b5d214fa4..4455eaab258 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DeprecationHttpIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DeprecationHttpIT.java @@ -78,7 +78,7 @@ public class DeprecationHttpIT extends HttpSmokeTestCase { * Attempts to do a scatter/gather request that expects unique responses per sub-request. */ @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/19222") - public void testUniqueDeprecationResponsesMergedTogether() throws IOException { + public void testUniqueDeprecationResponsesMergedTogether() throws Exception { final String[] indices = new String[randomIntBetween(2, 5)]; // add at least one document for each index @@ -99,35 +99,31 @@ public class DeprecationHttpIT extends HttpSmokeTestCase { final String commaSeparatedIndices = Stream.of(indices).collect(Collectors.joining(",")); - final String body = - "{\"query\":{\"bool\":{\"filter\":[{\"" + TestDeprecatedQueryBuilder.NAME + "\":{}}]}}}"; + final String body = "{\"query\":{\"bool\":{\"filter\":[{\"" + TestDeprecatedQueryBuilder.NAME + "\":{}}]}}}"; // trigger all index deprecations - try (Response response = getRestClient().performRequest("GET", - "/" + commaSeparatedIndices + "/_search", - Collections.emptyMap(), - new StringEntity(body, RestClient.JSON_CONTENT_TYPE))) { - assertThat(response.getStatusLine().getStatusCode(), equalTo(OK.getStatus())); + Response response = getRestClient().performRequest("GET", "/" + commaSeparatedIndices + "/_search", + Collections.emptyMap(), new StringEntity(body, RestClient.JSON_CONTENT_TYPE)); + assertThat(response.getStatusLine().getStatusCode(), equalTo(OK.getStatus())); - final List deprecatedWarnings = getWarningHeaders(response.getHeaders()); - final List> headerMatchers = new ArrayList<>(indices.length); + final List deprecatedWarnings = getWarningHeaders(response.getHeaders()); + final List> headerMatchers = new ArrayList<>(indices.length); - for (String index : indices) { - headerMatchers.add(containsString(LoggerMessageFormat.format("[{}] index", (Object)index))); - } + for (String index : indices) { + headerMatchers.add(containsString(LoggerMessageFormat.format("[{}] index", (Object)index))); + } - assertThat(deprecatedWarnings, hasSize(headerMatchers.size())); - for (Matcher headerMatcher : headerMatchers) { - assertThat(deprecatedWarnings, hasItem(headerMatcher)); - } + assertThat(deprecatedWarnings, hasSize(headerMatchers.size())); + for (Matcher headerMatcher : headerMatchers) { + assertThat(deprecatedWarnings, hasItem(headerMatcher)); } } - public void testDeprecationWarningsAppearInHeaders() throws IOException { + public void testDeprecationWarningsAppearInHeaders() throws Exception { doTestDeprecationWarningsAppearInHeaders(); } - public void testDeprecationHeadersDoNotGetStuck() throws IOException { + public void testDeprecationHeadersDoNotGetStuck() throws Exception { doTestDeprecationWarningsAppearInHeaders(); doTestDeprecationWarningsAppearInHeaders(); if (rarely()) { @@ -140,7 +136,7 @@ public class DeprecationHttpIT extends HttpSmokeTestCase { *

* Re-running this back-to-back helps to ensure that warnings are not being maintained across requests. */ - private void doTestDeprecationWarningsAppearInHeaders() throws IOException { + private void doTestDeprecationWarningsAppearInHeaders() throws Exception { final boolean useDeprecatedField = randomBoolean(); final boolean useNonDeprecatedSetting = randomBoolean(); @@ -159,29 +155,26 @@ public class DeprecationHttpIT extends HttpSmokeTestCase { Collections.shuffle(settings, random()); // trigger all deprecations - try (Response response = getRestClient().performRequest("GET", - "/_test_cluster/deprecated_settings", - Collections.emptyMap(), - buildSettingsRequest(settings, useDeprecatedField))) { - assertThat(response.getStatusLine().getStatusCode(), equalTo(OK.getStatus())); + Response response = getRestClient().performRequest("GET", "/_test_cluster/deprecated_settings", + Collections.emptyMap(), buildSettingsRequest(settings, useDeprecatedField)); + assertThat(response.getStatusLine().getStatusCode(), equalTo(OK.getStatus())); - final List deprecatedWarnings = getWarningHeaders(response.getHeaders()); - final List> headerMatchers = new ArrayList<>(4); + final List deprecatedWarnings = getWarningHeaders(response.getHeaders()); + final List> headerMatchers = new ArrayList<>(4); - headerMatchers.add(equalTo(TestDeprecationHeaderRestAction.DEPRECATED_ENDPOINT)); - if (useDeprecatedField) { - headerMatchers.add(equalTo(TestDeprecationHeaderRestAction.DEPRECATED_USAGE)); - } - for (Setting setting : settings) { - if (setting.isDeprecated()) { - headerMatchers.add(containsString(LoggerMessageFormat.format("[{}] setting was deprecated", (Object)setting.getKey()))); - } + headerMatchers.add(equalTo(TestDeprecationHeaderRestAction.DEPRECATED_ENDPOINT)); + if (useDeprecatedField) { + headerMatchers.add(equalTo(TestDeprecationHeaderRestAction.DEPRECATED_USAGE)); + } + for (Setting setting : settings) { + if (setting.isDeprecated()) { + headerMatchers.add(containsString(LoggerMessageFormat.format("[{}] setting was deprecated", (Object)setting.getKey()))); } + } - assertThat(deprecatedWarnings, hasSize(headerMatchers.size())); - for (Matcher headerMatcher : headerMatchers) { - assertThat(deprecatedWarnings, hasItem(headerMatcher)); - } + assertThat(deprecatedWarnings, hasSize(headerMatchers.size())); + for (Matcher headerMatcher : headerMatchers) { + assertThat(deprecatedWarnings, hasItem(headerMatcher)); } } diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsDisabledIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsDisabledIT.java index feca7cd1d5f..5e53599b934 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsDisabledIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsDisabledIT.java @@ -19,11 +19,11 @@ package org.elasticsearch.http; +import org.apache.http.util.EntityUtils; import org.elasticsearch.client.Response; import org.elasticsearch.client.ResponseException; import org.elasticsearch.common.network.NetworkModule; import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.http.HttpTransportSettings; import org.elasticsearch.test.ESIntegTestCase; import org.elasticsearch.test.ESIntegTestCase.ClusterScope; import org.elasticsearch.test.ESIntegTestCase.Scope; @@ -59,7 +59,7 @@ public class DetailedErrorsDisabledIT extends HttpSmokeTestCase { } catch(ResponseException e) { Response response = e.getResponse(); assertThat(response.getHeader("Content-Type"), is("application/json")); - assertThat(e.getResponseBody(), is("{\"error\":\"error traces in responses are disabled.\"}")); + assertThat(EntityUtils.toString(e.getResponse().getEntity()), is("{\"error\":\"error traces in responses are disabled.\"}")); assertThat(response.getStatusLine().getStatusCode(), is(400)); } } diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsEnabledIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsEnabledIT.java index daabb1bc70d..fb26e59a1a5 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsEnabledIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/DetailedErrorsEnabledIT.java @@ -19,13 +19,10 @@ package org.elasticsearch.http; +import org.apache.http.util.EntityUtils; import org.elasticsearch.client.Response; import org.elasticsearch.client.ResponseException; -import org.elasticsearch.common.network.NetworkModule; -import org.elasticsearch.common.settings.Settings; import org.elasticsearch.test.ESIntegTestCase; -import org.elasticsearch.test.ESIntegTestCase.ClusterScope; -import org.elasticsearch.test.ESIntegTestCase.Scope; import java.util.Collections; @@ -44,7 +41,8 @@ public class DetailedErrorsEnabledIT extends HttpSmokeTestCase { } catch(ResponseException e) { Response response = e.getResponse(); assertThat(response.getHeader("Content-Type"), containsString("application/json")); - assertThat(e.getResponseBody(), containsString("\"stack_trace\":\"[Validation Failed: 1: index / indices is missing;]; " + + assertThat(EntityUtils.toString(response.getEntity()), + containsString("\"stack_trace\":\"[Validation Failed: 1: index / indices is missing;]; " + "nested: ActionRequestValidationException[Validation Failed: 1:")); } @@ -54,7 +52,8 @@ public class DetailedErrorsEnabledIT extends HttpSmokeTestCase { } catch(ResponseException e) { Response response = e.getResponse(); assertThat(response.getHeader("Content-Type"), containsString("application/json")); - assertThat(e.getResponseBody(), not(containsString("\"stack_trace\":\"[Validation Failed: 1: index / indices is missing;]; " + assertThat(EntityUtils.toString(response.getEntity()), + not(containsString("\"stack_trace\":\"[Validation Failed: 1: index / indices is missing;]; " + "nested: ActionRequestValidationException[Validation Failed: 1:"))); } } diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/HttpCompressionIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/HttpCompressionIT.java index f08bb2b4a9e..ca637a78555 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/HttpCompressionIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/HttpCompressionIT.java @@ -18,19 +18,13 @@ */ package org.elasticsearch.http; -import org.apache.http.Header; -import org.apache.http.HttpException; import org.apache.http.HttpHeaders; -import org.apache.http.HttpResponseInterceptor; import org.apache.http.entity.StringEntity; -import org.apache.http.impl.client.HttpClientBuilder; import org.apache.http.message.BasicHeader; -import org.apache.http.protocol.HttpContext; import org.elasticsearch.client.Response; import org.elasticsearch.client.RestClient; import org.elasticsearch.test.ESIntegTestCase; -import java.io.IOException; import java.util.Collections; public class HttpCompressionIT extends ESIntegTestCase { @@ -50,99 +44,23 @@ public class HttpCompressionIT extends ESIntegTestCase { public void testCompressesResponseIfRequested() throws Exception { ensureGreen(); - // we need to intercept early, otherwise internal logic in HttpClient will just remove the header and we cannot verify it - ContentEncodingHeaderExtractor headerExtractor = new ContentEncodingHeaderExtractor(); - try (RestClient client = createRestClient(new ContentEncodingHeaderExtractorConfigCallback(headerExtractor))) { - try (Response response = client.performRequest("GET", "/", new BasicHeader(HttpHeaders.ACCEPT_ENCODING, GZIP_ENCODING))) { - assertEquals(200, response.getStatusLine().getStatusCode()); - assertTrue(headerExtractor.hasContentEncodingHeader()); - assertEquals(GZIP_ENCODING, headerExtractor.getContentEncodingHeader().getValue()); - } + try (RestClient client = getRestClient()) { + Response response = client.performRequest("GET", "/", new BasicHeader(HttpHeaders.ACCEPT_ENCODING, GZIP_ENCODING)); + assertEquals(200, response.getStatusLine().getStatusCode()); + assertEquals(GZIP_ENCODING, response.getHeader(HttpHeaders.CONTENT_ENCODING)); } } public void testUncompressedResponseByDefault() throws Exception { ensureGreen(); - ContentEncodingHeaderExtractor headerExtractor = new ContentEncodingHeaderExtractor(); - try (RestClient client = createRestClient(new NoContentCompressionConfigCallback(headerExtractor))) { - try (Response response = client.performRequest("GET", "/")) { - assertEquals(200, response.getStatusLine().getStatusCode()); - assertFalse(headerExtractor.hasContentEncodingHeader()); - } - } - } + try (RestClient client = getRestClient()) { + Response response = client.performRequest("GET", "/"); + assertEquals(200, response.getStatusLine().getStatusCode()); + assertNull(response.getHeader(HttpHeaders.CONTENT_ENCODING)); - public void testCanInterpretUncompressedRequest() throws Exception { - ensureGreen(); - ContentEncodingHeaderExtractor headerExtractor = new ContentEncodingHeaderExtractor(); - // this disable content compression in both directions (request and response) - try (RestClient client = createRestClient(new NoContentCompressionConfigCallback(headerExtractor))) { - try (Response response = client.performRequest("POST", "/company/employees/1", - Collections.emptyMap(), SAMPLE_DOCUMENT)) { - assertEquals(201, response.getStatusLine().getStatusCode()); - assertFalse(headerExtractor.hasContentEncodingHeader()); - } - } - } - - public void testCanInterpretCompressedRequest() throws Exception { - ensureGreen(); - ContentEncodingHeaderExtractor headerExtractor = new ContentEncodingHeaderExtractor(); - // we don't call #disableContentCompression() hence the client will send the content compressed - try (RestClient client = createRestClient(new ContentEncodingHeaderExtractorConfigCallback(headerExtractor))) { - try (Response response = client.performRequest("POST", "/company/employees/2", - Collections.emptyMap(), SAMPLE_DOCUMENT)) { - assertEquals(201, response.getStatusLine().getStatusCode()); - assertEquals(GZIP_ENCODING, headerExtractor.getContentEncodingHeader().getValue()); - } - } - } - - private static class ContentEncodingHeaderExtractor implements HttpResponseInterceptor { - private Header contentEncodingHeader; - - @Override - public void process(org.apache.http.HttpResponse response, HttpContext context) throws HttpException, IOException { - final Header[] headers = response.getHeaders(HttpHeaders.CONTENT_ENCODING); - if (headers.length == 1) { - this.contentEncodingHeader = headers[0]; - } else if (headers.length > 1) { - throw new AssertionError("Expected none or one content encoding header but got " + headers.length + " headers."); - } - } - - public boolean hasContentEncodingHeader() { - return contentEncodingHeader != null; - } - - public Header getContentEncodingHeader() { - return contentEncodingHeader; - } - } - - private static class NoContentCompressionConfigCallback extends ContentEncodingHeaderExtractorConfigCallback { - NoContentCompressionConfigCallback(ContentEncodingHeaderExtractor contentEncodingHeaderExtractor) { - super(contentEncodingHeaderExtractor); - } - - @Override - public void customizeHttpClient(HttpClientBuilder httpClientBuilder) { - super.customizeHttpClient(httpClientBuilder); - httpClientBuilder.disableContentCompression(); - } - } - - private static class ContentEncodingHeaderExtractorConfigCallback implements RestClient.HttpClientConfigCallback { - - private final ContentEncodingHeaderExtractor contentEncodingHeaderExtractor; - - ContentEncodingHeaderExtractorConfigCallback(ContentEncodingHeaderExtractor contentEncodingHeaderExtractor) { - this.contentEncodingHeaderExtractor = contentEncodingHeaderExtractor; - } - - @Override - public void customizeHttpClient(HttpClientBuilder httpClientBuilder) { - httpClientBuilder.addInterceptorFirst(contentEncodingHeaderExtractor); + response = client.performRequest("POST", "/company/employees/1", Collections.emptyMap(), SAMPLE_DOCUMENT); + assertEquals(201, response.getStatusLine().getStatusCode()); + assertNull(response.getHeader(HttpHeaders.CONTENT_ENCODING)); } } } diff --git a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ResponseHeaderPluginIT.java b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ResponseHeaderPluginIT.java index 482edc36702..037549ada06 100644 --- a/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ResponseHeaderPluginIT.java +++ b/qa/smoke-test-http/src/test/java/org/elasticsearch/http/ResponseHeaderPluginIT.java @@ -67,9 +67,8 @@ public class ResponseHeaderPluginIT extends HttpSmokeTestCase { assertThat(response.getHeader("Secret"), equalTo("required")); } - try (Response authResponse = getRestClient().performRequest("GET", "/_protected", new BasicHeader("Secret", "password"))) { - assertThat(authResponse.getStatusLine().getStatusCode(), equalTo(200)); - assertThat(authResponse.getHeader("Secret"), equalTo("granted")); - } + Response authResponse = getRestClient().performRequest("GET", "/_protected", new BasicHeader("Secret", "password")); + assertThat(authResponse.getStatusLine().getStatusCode(), equalTo(200)); + assertThat(authResponse.getHeader("Secret"), equalTo("granted")); } } diff --git a/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java index e8895aa90db..2df619a3a11 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java @@ -299,7 +299,7 @@ public abstract class ESRestTestCase extends ESTestCase { * other tests. */ @After - public void logIfThereAreRunningTasks() throws InterruptedException, IOException { + public void logIfThereAreRunningTasks() throws Exception { RestTestResponse tasks = adminExecutionContext.callApi("tasks.list", emptyMap(), emptyList(), emptyMap()); Set runningTasks = runningTasks(tasks); // Ignore the task list API - it doens't count against us @@ -341,7 +341,7 @@ public abstract class ESRestTestCase extends ESTestCase { } @Before - public void reset() throws IOException { + public void reset() throws Exception { // admin context must be available for @After always, regardless of whether the test was blacklisted adminExecutionContext.initClient(clusterUrls, restAdminSettings()); adminExecutionContext.clear(); @@ -378,7 +378,7 @@ public abstract class ESRestTestCase extends ESTestCase { return messageBuilder.toString(); } - public void test() throws IOException { + public void test() throws Exception { //let's check that there is something to run, otherwise there might be a problem with the test section if (testCandidate.getTestSection().getExecutableSections().size() == 0) { throw new IllegalArgumentException("No executable sections loaded for [" + testCandidate.getTestPath() + "]"); diff --git a/test/framework/src/main/java/org/elasticsearch/test/rest/RestTestExecutionContext.java b/test/framework/src/main/java/org/elasticsearch/test/rest/RestTestExecutionContext.java index d7295e1dca7..cde95ff9812 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/rest/RestTestExecutionContext.java +++ b/test/framework/src/main/java/org/elasticsearch/test/rest/RestTestExecutionContext.java @@ -62,7 +62,7 @@ public class RestTestExecutionContext implements Closeable { * Saves the obtained response in the execution context. */ public RestTestResponse callApi(String apiName, Map params, List> bodies, - Map headers) throws IOException { + Map headers) throws Exception { //makes a copy of the parameters before modifying them for this specific request HashMap requestParams = new HashMap<>(params); for (Map.Entry entry : requestParams.entrySet()) { @@ -79,7 +79,7 @@ public class RestTestExecutionContext implements Closeable { stash.stashValue("body", response.getBody()); return response; } catch(ResponseException e) { - response = new RestTestResponse(e); + response = new RestTestResponse(e.getResponse()); throw e; } } @@ -105,7 +105,7 @@ public class RestTestExecutionContext implements Closeable { } private RestTestResponse callApiInternal(String apiName, Map params, String body, Map headers) - throws IOException { + throws Exception { return restTestClient.callApi(apiName, params, body, headers); } @@ -119,7 +119,7 @@ public class RestTestExecutionContext implements Closeable { /** * Creates the embedded REST client when needed. Needs to be called before each test. */ - public void initClient(URL[] urls, Settings settings) throws IOException { + public void initClient(URL[] urls, Settings settings) throws Exception { if (restTestClient == null) { restTestClient = new RestTestClient(restSpec, settings, urls); } diff --git a/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestClient.java b/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestClient.java index c97ca7cd2fd..40fae122a5e 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestClient.java +++ b/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestClient.java @@ -22,16 +22,15 @@ import com.carrotsearch.randomizedtesting.RandomizedTest; import org.apache.http.Header; import org.apache.http.HttpEntity; import org.apache.http.HttpHost; -import org.apache.http.conn.ssl.SSLConnectionSocketFactory; import org.apache.http.entity.StringEntity; import org.apache.http.message.BasicHeader; +import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy; import org.apache.http.ssl.SSLContexts; import org.apache.lucene.util.IOUtils; import org.elasticsearch.Version; import org.elasticsearch.client.Response; import org.elasticsearch.client.ResponseException; import org.elasticsearch.client.RestClient; -import org.elasticsearch.client.SSLSocketFactoryHttpConfigCallback; import org.elasticsearch.common.Strings; import org.elasticsearch.common.io.PathUtils; import org.elasticsearch.common.logging.ESLogger; @@ -83,7 +82,7 @@ public class RestTestClient implements Closeable { private final RestClient restClient; private final Version esVersion; - public RestTestClient(RestSpec restSpec, Settings settings, URL[] urls) throws IOException { + public RestTestClient(RestSpec restSpec, Settings settings, URL[] urls) throws Exception { assert urls.length > 0; this.restSpec = restSpec; this.restClient = createRestClient(urls, settings); @@ -91,7 +90,7 @@ public class RestTestClient implements Closeable { logger.info("REST client initialized {}, elasticsearch version: [{}]", urls, esVersion); } - private Version readAndCheckVersion(URL[] urls) throws IOException { + private Version readAndCheckVersion(URL[] urls) throws Exception { RestApi restApi = restApi("info"); assert restApi.getPaths().size() == 1; assert restApi.getMethods().size() == 1; @@ -126,7 +125,7 @@ public class RestTestClient implements Closeable { * Calls an api with the provided parameters and body */ public RestTestResponse callApi(String apiName, Map params, String body, Map headers) - throws IOException { + throws Exception { if ("raw".equals(apiName)) { // Raw requests are bit simpler.... @@ -247,7 +246,7 @@ public class RestTestClient implements Closeable { return new RestTestResponse(response); } catch(ResponseException e) { if (ignores.contains(e.getResponse().getStatusLine().getStatusCode())) { - return new RestTestResponse(e); + return new RestTestResponse(e.getResponse()); } throw e; } @@ -287,8 +286,8 @@ public class RestTestClient implements Closeable { keyStore.load(is, keystorePass.toCharArray()); } SSLContext sslcontext = SSLContexts.custom().loadTrustMaterial(keyStore, null).build(); - SSLConnectionSocketFactory sslConnectionSocketFactory = new SSLConnectionSocketFactory(sslcontext); - builder.setHttpClientConfigCallback(new SSLSocketFactoryHttpConfigCallback(sslConnectionSocketFactory)); + SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslcontext); + builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setSSLStrategy(sessionStrategy)); } catch (KeyStoreException|NoSuchAlgorithmException|KeyManagementException|CertificateException e) { throw new RuntimeException(e); } diff --git a/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestResponse.java b/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestResponse.java index 4644b87b8e7..27e1abdbdba 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestResponse.java +++ b/test/framework/src/main/java/org/elasticsearch/test/rest/client/RestTestResponse.java @@ -20,9 +20,7 @@ package org.elasticsearch.test.rest.client; import org.apache.http.client.methods.HttpHead; import org.apache.http.util.EntityUtils; -import org.apache.lucene.util.IOUtils; import org.elasticsearch.client.Response; -import org.elasticsearch.client.ResponseException; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.test.rest.ObjectPath; import org.elasticsearch.test.rest.Stash; @@ -48,8 +46,6 @@ public class RestTestResponse { } catch (IOException e) { EntityUtils.consumeQuietly(response.getEntity()); throw new RuntimeException(e); - } finally { - IOUtils.closeWhileHandlingException(response); } } else { this.body = null; @@ -57,12 +53,6 @@ public class RestTestResponse { parseResponseBody(); } - public RestTestResponse(ResponseException responseException) throws IOException { - this.response = responseException.getResponse(); - this.body = responseException.getResponseBody(); - parseResponseBody(); - } - private void parseResponseBody() throws IOException { if (body != null) { String contentType = response.getHeader("Content-Type"); diff --git a/test/framework/src/main/java/org/elasticsearch/test/rest/section/DoSection.java b/test/framework/src/main/java/org/elasticsearch/test/rest/section/DoSection.java index 2547d6becea..8b242c53986 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/rest/section/DoSection.java +++ b/test/framework/src/main/java/org/elasticsearch/test/rest/section/DoSection.java @@ -79,7 +79,7 @@ public class DoSection implements ExecutableSection { } @Override - public void execute(RestTestExecutionContext executionContext) throws IOException { + public void execute(RestTestExecutionContext executionContext) throws Exception { if ("param".equals(catchParam)) { //client should throw validation error before sending request @@ -103,7 +103,7 @@ public class DoSection implements ExecutableSection { fail(formatStatusCodeMessage(restTestResponse, catchStatusCode)); } } catch(ResponseException e) { - RestTestResponse restTestResponse = new RestTestResponse(e); + RestTestResponse restTestResponse = new RestTestResponse(e.getResponse()); if (!Strings.hasLength(catchParam)) { fail(formatStatusCodeMessage(restTestResponse, "2xx")); } else if (catches.containsKey(catchParam)) { diff --git a/test/framework/src/main/java/org/elasticsearch/test/rest/section/ExecutableSection.java b/test/framework/src/main/java/org/elasticsearch/test/rest/section/ExecutableSection.java index 669d82cdd78..ece972b77f9 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/rest/section/ExecutableSection.java +++ b/test/framework/src/main/java/org/elasticsearch/test/rest/section/ExecutableSection.java @@ -20,8 +20,6 @@ package org.elasticsearch.test.rest.section; import org.elasticsearch.test.rest.RestTestExecutionContext; -import java.io.IOException; - /** * Represents a test fragment that can be executed (e.g. api call, assertion) */ @@ -30,5 +28,5 @@ public interface ExecutableSection { /** * Executes the section passing in the execution context */ - void execute(RestTestExecutionContext executionContext) throws IOException; + void execute(RestTestExecutionContext executionContext) throws Exception; }