Rest client: introduce async performRequest method and use async client under the hood for sync requests too

The new method accepts the usual parameters (method, endpoint, params, entity and headers) plus a response listener and an async response consumer. Shortcut methods are also added that don't require params, entity and the async response consumer optional.

There are a few relevant api changes as a consequence of the move to async client that affect sync methods:
- Response doesn't implement Closeable anymore, responses don't need to be closed
- performRequest throws Exception rather than just IOException, as that is the the exception that we get from the FutureCallback#failed method in the async http client
- ssl configuration is a bit simpler, one only needs to call setSSLStrategy from a custom HttpClientConfigCallback, that doesn't end up overridng any other default around connection pooling (it used to happen with the sync client and make ssl configuration more complex)

Relates to #19055

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);
         }

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();
-    }
 }

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
+ * Holds the response that was returned.
- * is expected to be closed and available only to read metadata like status line, request line, response headers.
  */
 public class ResponseException extends IOException {
 
     private Response response;
-    private final String responseBody;
 
-    ResponseException(Response response, String responseBody) throws IOException {
+    ResponseException(Response response) throws IOException {
-        super(buildMessage(response,responseBody));
+        super(buildMessage(response));
         this.response = response;
-        this.responseBody = responseBody;
     }
 
-    private static String buildMessage(Response response, String responseBody) {
+    private static String buildMessage(Response response) {
-        String message = response.getRequestLine().getMethod() + " " + response.getHost() + response.getRequestLine().getUri()
+        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;
-    }
 }

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
+ * Listener to be provided when calling async performRequest methods provided by {@link RestClient}.
- * org.apache.http.impl.execchain.HttpResponseProxy is not public.
+ * 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 {
+     * Method invoked if the request failed. There are two main categories of failures: connection failures (usually
-        //nothing to close
+     * {@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);
 }

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.impl.nio.client.CloseableHttpAsyncClient;
-import org.apache.http.util.EntityUtils;
+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;
+    private final CloseableHttpAsyncClient client;
-    //we don't rely on default headers supported by HttpClient as those cannot be replaced, plus it would get hairy
+    //we don't rely on default headers supported by HttpAsyncClient as those cannot be replaced
-    //when we create the HttpClient instance on our own as there would be two different ways to set the default headers.
     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<HttpHost, DeadHostState> 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 {
+    public Response performRequest(String method, String endpoint, Header... headers) throws Exception {
-        return performRequest(method, endpoint, Collections.<String, String>emptyMap(), null, headers);
+        return performRequest(method, endpoint, Collections.<String, String>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<String, String> params, Header... headers) throws IOException {
+    public Response performRequest(String method, String endpoint, Map<String, String> params, Header... headers) throws Exception {
-        return performRequest(method, endpoint, params, null, headers);
+        return performRequest(method, endpoint, params, (HttpEntity)null, headers);
     }
 
     /**
-     * Sends a request to the elasticsearch cluster that the current client points to.
+     * Sends a request to the elasticsearch cluster that the current client points to. Blocks till the request is completed and returns
-     * Selects a host out of the provided ones in a round-robin fashion. Failing hosts are marked dead and retried after a certain
+     * its response of fails by throwing an exception. Selects a host out of the provided ones in a round-robin fashion. Failing hosts
-     * amount of time (minimum 1 minute, maximum 30 minutes), depending on how many times they previously failed (the more failures,
+     * are marked dead and retried after a certain amount of time (minimum 1 minute, maximum 30 minutes), depending on how many times
-     * the later they will be retried). In case of failures all of the alive nodes (or dead nodes that deserve a retry) are retried
+     * they previously failed (the more failures, the later they will be retried). In case of failures all of the alive nodes (or dead
-     * till one responds or none of them does, in which case an {@link IOException} will be thrown.
+     * 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<String, String> params,
-                                   HttpEntity entity, Header... headers) throws IOException {
+                                   HttpEntity entity, Header... headers) throws Exception {
+        HttpAsyncResponseConsumer<HttpResponse> 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.<String, String>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<String, String> 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<String, String> params,
+                               HttpEntity entity, ResponseListener responseListener, Header... headers) {
+        HttpAsyncResponseConsumer<HttpResponse> 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<String, String> params,
+                               HttpEntity entity, HttpAsyncResponseConsumer<HttpResponse> 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
+        FailureTrackingListener failureTrackingListener = new FailureTrackingListener(responseListener);
-        long retryTimeoutMillis = Math.round(this.maxRetryTimeoutMillis / (float)100 * 98);
-        IOException lastSeenException = null;
         long startTime = System.nanoTime();
-        for (HttpHost host : nextHost()) {
+        performRequest(startTime, nextHost().iterator(), request, responseConsumer, failureTrackingListener);
-            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;
-                }
-                //also reset the request to make it reusable for the next attempt
-                request.reset();
     }
 
-            CloseableHttpResponse httpResponse;
+    private void performRequest(final long startTime, final Iterator<HttpHost> hosts, final HttpRequestBase request,
+                                final HttpAsyncResponseConsumer<HttpResponse> 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<HttpResponse>() {
+            @Override
+            public void completed(HttpResponse httpResponse) {
                 try {
-                httpResponse = client.execute(host, request);
+                    RequestLogger.logResponse(logger, request, host, httpResponse);
-            } catch(IOException e) {
+                    int statusCode = httpResponse.getStatusLine().getStatusCode();
-                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 (isSuccessfulResponse(request.getMethod(), statusCode)) {
-            if (statusCode < 300 || (request.getMethod().equals(HttpHead.METHOD_NAME) && statusCode == 404) ) {
-                RequestLogger.logResponse(logger, request, host, httpResponse);
                         onResponse(host);
-                return response;
+                        listener.onSuccess(response);
-            }
-            RequestLogger.logResponse(logger, request, host, httpResponse);
-            String responseBody;
-            try {
-                if (response.getEntity() == null) {
-                    responseBody = null;
                     } else {
-                    responseBody = EntityUtils.toString(response.getEntity());
+                        ResponseException responseException = new ResponseException(response);
-                }
+                        if (mustRetry(statusCode)) {
-            } 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
                             onFailure(host);
-                    break;
+                            retryIfPossible(responseException, hosts, request);
-                default:
+                        } else {
                             //mark host alive and don't retry, as the error should be a request problem
                             onResponse(host);
-                    throw lastSeenException;
+                            listener.onDefinitiveFailure(responseException);
                         }
                     }
-        //we get here only when we tried all nodes and they all failed
+                } catch(Exception e) {
-        assert lastSeenException != null;
+                    listener.onDefinitiveFailure(e);
-        throw lastSeenException;
+                }
+            }
+
+            @Override
+            public void failed(Exception failure) {
+                try {
+                    RequestLogger.logFailedRequest(logger, request, host, failure);
+                    onFailure(host);
+                    retryIfPossible(failure, hosts, request);
+                } catch(Exception e) {
+                    listener.onDefinitiveFailure(e);
+                }
+            }
+
+            private void retryIfPossible(Exception exception, Iterator<HttpHost> 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.
+         * Sets the default request headers, which will be sent along with each 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
          */
         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}.
+         * Allows to customize the {@link CloseableHttpAsyncClient} being created and used by the {@link RestClient}.
-         * It is common to customzie the default {@link org.apache.http.client.CredentialsProvider} through this method,
+         * Commonly used to customize the default {@link org.apache.http.client.CredentialsProvider} for authentication
-         * without losing any other useful default value that the {@link RestClient.Builder} internally sets.
+         * or the {@link SchemeIOSessionStrategy} for communication through ssl without losing any other useful default
-         * Also useful to setup ssl through {@link SSLSocketFactoryHttpConfigCallback}.
+         * 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) {
 
         }
     }

client/rest/src/main/java/org/elasticsearch/client/SSLSocketFactoryHttpConfigCallback.java

@@ -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<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>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);
-    }
-}

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) {
                 }
             });
         }

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<String> 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,
+            try {
-                    Collections.<String, String>emptyMap(), null, headers)) {
+                esResponse = restClient.performRequest(method, "/" + statusCode, Collections.<String, String>emptyMap(),
-                esResponse = response;
+                        (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,
+        Response esResponse;
-                Collections.<String, String>emptyMap(), entity)) {
+        try {
-            responseBody = EntityUtils.toString(response.getEntity());
+            esResponse = restClient.performRequest(method, "/" + statusCode, Collections.<String, String>emptyMap(), entity);
-            esResponse = response;
         } catch(ResponseException e) {
-            responseBody = e.getResponseBody();
             esResponse = e.getResponse();
         }
         assertEquals(statusCode, esResponse.getStatusLine().getStatusCode());
-        assertEquals(requestBody, responseBody);
+        assertEquals(requestBody, EntityUtils.toString(esResponse.getEntity()));
     }
 }

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);
+        CloseableHttpAsyncClient httpClient = mock(CloseableHttpAsyncClient.class);
-        when(httpClient.execute(any(HttpHost.class), any(HttpRequest.class))).thenAnswer(new Answer<CloseableHttpResponse>() {
+        when(httpClient.<HttpResponse>execute(any(HttpAsyncRequestProducer.class), any(HttpAsyncResponseConsumer.class),
+                any(FutureCallback.class))).thenAnswer(new Answer<Future<HttpResponse>>() {
             @Override
-            public CloseableHttpResponse answer(InvocationOnMock invocationOnMock) throws Throwable {
+            public Future<HttpResponse> answer(InvocationOnMock invocationOnMock) throws Throwable {
-                HttpHost httpHost = (HttpHost) invocationOnMock.getArguments()[0];
+                HttpAsyncRequestProducer requestProducer = (HttpAsyncRequestProducer) invocationOnMock.getArguments()[0];
-                HttpUriRequest request = (HttpUriRequest) invocationOnMock.getArguments()[1];
+                HttpUriRequest request = (HttpUriRequest)requestProducer.generateRequest();
+                HttpHost httpHost = requestProducer.getTarget();
+                @SuppressWarnings("unchecked")
+                FutureCallback<HttpResponse> futureCallback = (FutureCallback<HttpResponse>) 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, "");
-                return new CloseableBasicHttpResponse(statusLine);
+                    futureCallback.completed(new BasicHttpResponse(statusLine));
+                }
+                return null;
             }
         });
-
         int numHosts = RandomInts.randomIntBetween(getRandom(), 2, 5);
         httpHosts = new HttpHost[numHosts];
         for (int i = 0; i < numHosts; i++) {
@@ -102,11 +112,10 @@ 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)) {
+                Response response = restClient.performRequest(randomHttpMethod(getRandom()), "/" + statusCode);
-                    assertThat(response.getStatusLine().getStatusCode(), equalTo(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());
         }
         failureListener.assertNotCalled();
@@ -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));
+                        assertEquals(404, response.getStatusLine().getStatusCode());
-                        assertThat(response.getStatusLine().getStatusCode(), equalTo(statusCode));
+                        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)) {
+                    try {
-                        response = esResponse;
+                        response = restClient.performRequest(randomHttpMethod(getRandom()), "/" + statusCode);
                     }
                     catch(ResponseException e) {
                         response = e.getResponse();

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);
+        httpClient = mock(CloseableHttpAsyncClient.class);
-        when(httpClient.execute(any(HttpHost.class), any(HttpRequest.class))).thenAnswer(new Answer<CloseableHttpResponse>() {
+        when(httpClient.<HttpResponse>execute(any(HttpAsyncRequestProducer.class), any(HttpAsyncResponseConsumer.class),
+                any(FutureCallback.class))).thenAnswer(new Answer<Future<HttpResponse>>() {
                     @Override
-            public CloseableHttpResponse answer(InvocationOnMock invocationOnMock) throws Throwable {
+                    public Future<HttpResponse> answer(InvocationOnMock invocationOnMock) throws Throwable {
-                HttpUriRequest request = (HttpUriRequest) invocationOnMock.getArguments()[1];
+                        HttpAsyncRequestProducer requestProducer = (HttpAsyncRequestProducer) invocationOnMock.getArguments()[0];
+                        @SuppressWarnings("unchecked")
+                        FutureCallback<HttpResponse> futureCallback = (FutureCallback<HttpResponse>) 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")) {
-                    throw new SocketTimeoutException();
+                            futureCallback.failed(new SocketTimeoutException());
                         } else if (request.getURI().getPath().equals("/coe")) {
-                    throw new ConnectTimeoutException();
+                            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);
+                            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());
+                                    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());
-                return httpResponse;
+                            futureCallback.completed(httpResponse);
+                        }
+                        return null;
                     }
                 });
+
+
         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<HttpUriRequest> requestArgumentCaptor = ArgumentCaptor.forClass(HttpUriRequest.class);
+        ArgumentCaptor<HttpAsyncRequestProducer> 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());
+            verify(httpClient, times(++times)).<HttpResponse>execute(requestArgumentCaptor.capture(),
-            HttpUriRequest actualRequest = requestArgumentCaptor.getValue();
+                    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,12 +269,10 @@ 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,
+                Response response = restClient.performRequest(method, "/" + okStatusCode, Collections.<String, String>emptyMap(), entity);
-                        Collections.<String, String>emptyMap(), entity)) {
                 assertThat(response.getStatusLine().getStatusCode(), equalTo(okStatusCode));
                 assertThat(EntityUtils.toString(response.getEntity()), equalTo(body));
             }
-            }
             for (int errorStatusCode : getAllErrorStatusCodes()) {
                 try {
                     restClient.performRequest(method, "/" + errorStatusCode, Collections.<String, String>emptyMap(), entity);
@@ -334,9 +349,8 @@ public class RestClientSingleHostTests extends RestClientTestCase {
 
             int statusCode = randomStatusCode(getRandom());
             Response esResponse;
-            try (Response response = restClient.performRequest(method, "/" + statusCode,
+            try {
-                    Collections.<String, String>emptyMap(), null, headers)) {
+                esResponse = restClient.performRequest(method, "/" + statusCode, headers);
-                esResponse = response;
             } 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<String, String> 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.<String, String>emptyMap(), headers);
             case 2:
-                return restClient.performRequest(method, endpoint, Collections.<String, String>emptyMap(), null, headers);
+                return restClient.performRequest(method, endpoint, Collections.<String, String>emptyMap(), (HttpEntity)null, headers);
             default:
                 throw new UnsupportedOperationException();
         }

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<HttpHost> hosts = new HashSet<>();
+    private volatile Set<HttpHost> hosts = new HashSet<>();
 
     @Override
-    public void onFailure(HttpHost host) throws IOException {
+    public void onFailure(HttpHost host) {
         hosts.add(host);
     }
 

client/sniffer/src/main/java/org/elasticsearch/client/sniff/HostsSniffer.java

@@ -61,11 +61,10 @@ public class HostsSniffer {
     /**
      * Calls the elasticsearch nodes info api, parses the response and returns all the found http hosts
      */
-    public List<HttpHost> sniffHosts() throws IOException {
+    public List<HttpHost> sniffHosts() throws Exception {
-        try (Response response = restClient.performRequest("get", "/_nodes/http", sniffRequestParams)) {
+        Response response = restClient.performRequest("get", "/_nodes/http", sniffRequestParams);
         return readHosts(response.getEntity());
     }
-    }
 
     private List<HttpHost> readHosts(HttpEntity entity) throws IOException {
         try (InputStream inputStream = entity.getContent()) {

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");
         }

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);

client/sniffer/src/test/java/org/elasticsearch/client/sniff/SniffOnFailureListenerTests.java

@@ -45,7 +45,7 @@ public class SniffOnFailureListenerTests extends RestClientTestCase {
             assertEquals("sniffer must not be null", e.getMessage());
         }
 
-        RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build();
+        try (RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build()) {
             try (Sniffer sniffer = Sniffer.builder(restClient, new MockHostsSniffer()).build()) {
                 listener.setSniffer(sniffer);
                 try {
@@ -57,4 +57,5 @@ public class SniffOnFailureListenerTests extends RestClientTestCase {
                 listener.onFailure(new HttpHost("localhost", 9200));
             }
         }
+    }
 }

modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteScrollableHitSource.java

@@ -214,11 +214,10 @@ 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)) {
+                    org.elasticsearch.client.Response response = restClient.performRequest(method, uri, params, entity);
                     InputStream markSupportedInputStream = new BufferedInputStream(response.getEntity().getContent());
                     listener.onResponse(markSupportedInputStream);
                 }
-                }
 
                 @Override
                 public void onFailure(Exception t) {

qa/smoke-test-http/src/test/java/org/elasticsearch/http/ContextAndHeaderTransportIT.java

@@ -220,10 +220,8 @@ public class ContextAndHeaderTransportIT extends HttpSmokeTestCase {
 
     public void testThatRelevantHttpHeadersBecomeRequestHeaders() throws Exception {
         final String IRRELEVANT_HEADER = "SomeIrrelevantHeader";
-
+        Response response = getRestClient().performRequest("GET", "/" + queryIndex + "/_search",
-        try (Response response = getRestClient().performRequest(
+                new BasicHeader(CUSTOM_HEADER, randomHeaderValue), new BasicHeader(IRRELEVANT_HEADER, randomHeaderValue));
-                "GET", "/" + queryIndex + "/_search",
-                new BasicHeader(CUSTOM_HEADER, randomHeaderValue), new BasicHeader(IRRELEVANT_HEADER, randomHeaderValue))) {
         assertThat(response.getStatusLine().getStatusCode(), equalTo(200));
         List<RequestAndHeaders> searchRequests = getRequests(SearchRequest.class);
         assertThat(searchRequests, hasSize(greaterThan(0)));
@@ -233,7 +231,6 @@ public class ContextAndHeaderTransportIT extends HttpSmokeTestCase {
             assertThat(requestAndHeaders.headers.containsKey(IRRELEVANT_HEADER), is(false));
         }
     }
-    }
 
     private  List<RequestAndHeaders> getRequests(Class<?> clazz) {
         List<RequestAndHeaders> results = new ArrayList<>();

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", "/",
+        Response response = getRestClient().performRequest("GET", "/",
-                new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue))) {
+                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", "/")) {
+        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());
     }
-    }
 }

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,17 +60,16 @@ public class CorsRegexIT extends HttpSmokeTestCase {
 
     public void testThatRegularExpressionWorksOnMatch() throws Exception {
         String corsValue = "http://localhost:9200";
-        try (Response response = getRestClient().performRequest("GET", "/",
+        Response response = getRestClient().performRequest("GET", "/",
-                new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue))) {
+                new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue));
         assertResponseWithOriginheader(response, corsValue);
-        }
+
         corsValue = "https://localhost:9200";
-        try (Response response = getRestClient().performRequest("GET", "/",
+        response = getRestClient().performRequest("GET", "/",
-                new BasicHeader("User-Agent", "Mozilla Bar"), new BasicHeader("Origin", corsValue));) {
+                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 {
         try {
@@ -85,28 +85,25 @@ public class CorsRegexIT extends HttpSmokeTestCase {
     }
 
     public void testThatSendingNoOriginHeaderReturnsNoAccessControlHeader() throws Exception {
-        try (Response response = getRestClient().performRequest("GET", "/", new BasicHeader("User-Agent", "Mozilla Bar"))) {
+        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", "/")) {
+        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"));) {
+                new BasicHeader(HttpHeaders.Names.ACCESS_CONTROL_REQUEST_METHOD, "GET"));
         assertResponseWithOriginheader(response, corsValue);
         assertNotNull(response.getHeader("Access-Control-Allow-Methods"));
     }
-    }
 
     public void testThatPreFlightRequestReturnsNullOnNonMatch() throws Exception {
         try {

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,14 +99,11 @@ public class DeprecationHttpIT extends HttpSmokeTestCase {
 
         final String commaSeparatedIndices = Stream.of(indices).collect(Collectors.joining(","));
 
-        final String body =
+        final String body = "{\"query\":{\"bool\":{\"filter\":[{\"" + TestDeprecatedQueryBuilder.NAME +  "\":{}}]}}}";
-            "{\"query\":{\"bool\":{\"filter\":[{\"" + TestDeprecatedQueryBuilder.NAME +  "\":{}}]}}}";
 
         // trigger all index deprecations
-        try (Response response = getRestClient().performRequest("GET",
+        Response response = getRestClient().performRequest("GET", "/" + commaSeparatedIndices + "/_search",
-                                                                "/" + commaSeparatedIndices + "/_search",
+                Collections.emptyMap(), new StringEntity(body, RestClient.JSON_CONTENT_TYPE));
-                                                                Collections.emptyMap(),
-                                                                new StringEntity(body, RestClient.JSON_CONTENT_TYPE))) {
         assertThat(response.getStatusLine().getStatusCode(), equalTo(OK.getStatus()));
 
         final List<String> deprecatedWarnings = getWarningHeaders(response.getHeaders());
@@ -121,13 +118,12 @@ public class DeprecationHttpIT extends HttpSmokeTestCase {
             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 {
      * <p>
      * 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,10 +155,8 @@ public class DeprecationHttpIT extends HttpSmokeTestCase {
         Collections.shuffle(settings, random());
 
         // trigger all deprecations
-        try (Response response = getRestClient().performRequest("GET",
+        Response response = getRestClient().performRequest("GET", "/_test_cluster/deprecated_settings",
-                                                                "/_test_cluster/deprecated_settings",
+                Collections.emptyMap(), buildSettingsRequest(settings, useDeprecatedField));
-                                                                Collections.emptyMap(),
-                                                                buildSettingsRequest(settings, useDeprecatedField))) {
         assertThat(response.getStatusLine().getStatusCode(), equalTo(OK.getStatus()));
 
         final List<String> deprecatedWarnings = getWarningHeaders(response.getHeaders());
@@ -183,7 +177,6 @@ public class DeprecationHttpIT extends HttpSmokeTestCase {
             assertThat(deprecatedWarnings, hasItem(headerMatcher));
         }
     }
-    }
 
     private List<String> getWarningHeaders(Header[] headers) {
         List<String> warnings = new ArrayList<>();

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));
         }
     }

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:")));
         }
     }

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
+        try (RestClient client = getRestClient()) {
-        ContentEncodingHeaderExtractor headerExtractor = new ContentEncodingHeaderExtractor();
+            Response response = client.performRequest("GET", "/", new BasicHeader(HttpHeaders.ACCEPT_ENCODING, GZIP_ENCODING));
-        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, response.getHeader(HttpHeaders.CONTENT_ENCODING));
-                assertEquals(GZIP_ENCODING, headerExtractor.getContentEncodingHeader().getValue());
-            }
         }
     }
 
     public void testUncompressedResponseByDefault() throws Exception {
         ensureGreen();
-        ContentEncodingHeaderExtractor headerExtractor = new ContentEncodingHeaderExtractor();
+        try (RestClient client = getRestClient()) {
-        try (RestClient client = createRestClient(new NoContentCompressionConfigCallback(headerExtractor))) {
+            Response response = client.performRequest("GET", "/");
-            try (Response response = client.performRequest("GET", "/")) {
             assertEquals(200, response.getStatusLine().getStatusCode());
-                assertFalse(headerExtractor.hasContentEncodingHeader());
+            assertNull(response.getHeader(HttpHeaders.CONTENT_ENCODING));
-            }
-        }
-    }
 
-    public void testCanInterpretUncompressedRequest() throws Exception {
+            response = client.performRequest("POST", "/company/employees/1", Collections.emptyMap(), SAMPLE_DOCUMENT);
-        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());
+            assertNull(response.getHeader(HttpHeaders.CONTENT_ENCODING));
-            }
-        }
-    }
-
-    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);
         }
     }
 }

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"))) {
+        Response authResponse = getRestClient().performRequest("GET", "/_protected", new BasicHeader("Secret", "password"));
         assertThat(authResponse.getStatusLine().getStatusCode(), equalTo(200));
         assertThat(authResponse.getHeader("Secret"), equalTo("granted"));
     }
-    }
 }

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<String> 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() + "]");

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<String, String> params, List<Map<String, Object>> bodies,
-                                    Map<String, String> headers) throws IOException  {
+                                    Map<String, String> headers) throws Exception  {
         //makes a copy of the parameters before modifying them for this specific request
         HashMap<String, String> requestParams = new HashMap<>(params);
         for (Map.Entry<String, String> 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<String, String> params, String body, Map<String, String> 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);
         }

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<String, String> params, String body, Map<String, String> 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);
+                SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslcontext);
-                builder.setHttpClientConfigCallback(new SSLSocketFactoryHttpConfigCallback(sslConnectionSocketFactory));
+                builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setSSLStrategy(sessionStrategy));
             } catch (KeyStoreException|NoSuchAlgorithmException|KeyManagementException|CertificateException e) {
                 throw new RuntimeException(e);
             }

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");

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)) {

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;
 }