HLRest: Allow caller to set per request options (#30490)

This modifies the high level rest client to allow calling code to
customize per request options for the bulk API. You do the actual
customization by passing a `RequestOptions` object to the API call
which is set on the `Request` that is generated by the high level
client. It also makes the `RequestOptions` a thing in the low level
rest client. For now that just means you use it to customize the
headers and the `httpAsyncResponseConsumerFactory` and we'll add
node selectors and per request timeouts in a follow up.

I only implemented this on the bulk API because it is the first one
in the list alphabetically and I wanted to keep the change small
enough to review. I'll convert the remaining APIs in a followup.

client/rest-high-level/src/main/java/org/elasticsearch/client/RestHighLevelClient.java

@@ -279,6 +279,17 @@ public class RestHighLevelClient implements Closeable {
      *
      * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html">Bulk API on elastic.co</a>
      */
+    public final BulkResponse bulk(BulkRequest bulkRequest, RequestOptions options) throws IOException {
+        return performRequestAndParseEntity(bulkRequest, RequestConverters::bulk, options, BulkResponse::fromXContent, emptySet());
+    }
+
+    /**
+     * Executes a bulk request using the Bulk API
+     *
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html">Bulk API on elastic.co</a>
+     * @deprecated Prefer {@link #bulk(BulkRequest, RequestOptions)}
+     */
+    @Deprecated
     public final BulkResponse bulk(BulkRequest bulkRequest, Header... headers) throws IOException {
         return performRequestAndParseEntity(bulkRequest, RequestConverters::bulk, BulkResponse::fromXContent, emptySet(), headers);
     }
@@ -288,6 +299,17 @@ public class RestHighLevelClient implements Closeable {
      *
      * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html">Bulk API on elastic.co</a>
      */
+    public final void bulkAsync(BulkRequest bulkRequest, RequestOptions options, ActionListener<BulkResponse> listener) {
+        performRequestAsyncAndParseEntity(bulkRequest, RequestConverters::bulk, options, BulkResponse::fromXContent, listener, emptySet());
+    }
+
+    /**
+     * Asynchronously executes a bulk request using the Bulk API
+     *
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html">Bulk API on elastic.co</a>
+     * @deprecated Prefer {@link #bulkAsync(BulkRequest, RequestOptions, ActionListener)}
+     */
+    @Deprecated
     public final void bulkAsync(BulkRequest bulkRequest, ActionListener<BulkResponse> listener, Header... headers) {
         performRequestAsyncAndParseEntity(bulkRequest, RequestConverters::bulk, BulkResponse::fromXContent, listener, emptySet(), headers);
     }
@@ -584,6 +606,7 @@ public class RestHighLevelClient implements Closeable {
             FieldCapabilitiesResponse::fromXContent, listener, emptySet(), headers);
     }
 
+    @Deprecated
     protected final <Req extends ActionRequest, Resp> Resp performRequestAndParseEntity(Req request,
                                                                             CheckedFunction<Req, Request, IOException> requestConverter,
                                                                             CheckedFunction<XContentParser, Resp, IOException> entityParser,
@@ -591,16 +614,34 @@ public class RestHighLevelClient implements Closeable {
         return performRequest(request, requestConverter, (response) -> parseEntity(response.getEntity(), entityParser), ignores, headers);
     }
 
+    protected final <Req extends ActionRequest, Resp> Resp performRequestAndParseEntity(Req request,
+                                                                            CheckedFunction<Req, Request, IOException> requestConverter,
+                                                                            RequestOptions options,
+                                                                            CheckedFunction<XContentParser, Resp, IOException> entityParser,
+                                                                            Set<Integer> ignores) throws IOException {
+        return performRequest(request, requestConverter, options,
+                response -> parseEntity(response.getEntity(), entityParser), ignores);
+    }
+
+    @Deprecated
     protected final <Req extends ActionRequest, Resp> Resp performRequest(Req request,
                                                           CheckedFunction<Req, Request, IOException> requestConverter,
                                                           CheckedFunction<Response, Resp, IOException> responseConverter,
                                                           Set<Integer> ignores, Header... headers) throws IOException {
+        return performRequest(request, requestConverter, optionsForHeaders(headers), responseConverter, ignores);
+    }
+
+    protected final <Req extends ActionRequest, Resp> Resp performRequest(Req request,
+                                                          CheckedFunction<Req, Request, IOException> requestConverter,
+                                                          RequestOptions options,
+                                                          CheckedFunction<Response, Resp, IOException> responseConverter,
+                                                          Set<Integer> ignores) throws IOException {
         ActionRequestValidationException validationException = request.validate();
         if (validationException != null) {
             throw validationException;
         }
         Request req = requestConverter.apply(request);
-        addHeaders(req, headers);
+        req.setOptions(options);
         Response response;
         try {
             response = client.performRequest(req);
@@ -626,6 +667,7 @@ public class RestHighLevelClient implements Closeable {
         }
     }
 
+    @Deprecated
     protected final <Req extends ActionRequest, Resp> void performRequestAsyncAndParseEntity(Req request,
                                                                  CheckedFunction<Req, Request, IOException> requestConverter,
                                                                  CheckedFunction<XContentParser, Resp, IOException> entityParser,
@@ -634,10 +676,28 @@ public class RestHighLevelClient implements Closeable {
                 listener, ignores, headers);
     }
 
+    protected final <Req extends ActionRequest, Resp> void performRequestAsyncAndParseEntity(Req request,
+                                                                 CheckedFunction<Req, Request, IOException> requestConverter,
+                                                                 RequestOptions options,
+                                                                 CheckedFunction<XContentParser, Resp, IOException> entityParser,
+                                                                 ActionListener<Resp> listener, Set<Integer> ignores) {
+        performRequestAsync(request, requestConverter, options,
+                response -> parseEntity(response.getEntity(), entityParser), listener, ignores);
+    }
+
+    @Deprecated
     protected final <Req extends ActionRequest, Resp> void performRequestAsync(Req request,
                                                                CheckedFunction<Req, Request, IOException> requestConverter,
                                                                CheckedFunction<Response, Resp, IOException> responseConverter,
                                                                ActionListener<Resp> listener, Set<Integer> ignores, Header... headers) {
+        performRequestAsync(request, requestConverter, optionsForHeaders(headers), responseConverter, listener, ignores);
+    }
+
+    protected final <Req extends ActionRequest, Resp> void performRequestAsync(Req request,
+                                                               CheckedFunction<Req, Request, IOException> requestConverter,
+                                                               RequestOptions options,
+                                                               CheckedFunction<Response, Resp, IOException> responseConverter,
+                                                               ActionListener<Resp> listener, Set<Integer> ignores) {
         ActionRequestValidationException validationException = request.validate();
         if (validationException != null) {
             listener.onFailure(validationException);
@@ -650,19 +710,12 @@ public class RestHighLevelClient implements Closeable {
             listener.onFailure(e);
             return;
         }
-        addHeaders(req, headers);
+        req.setOptions(options);
 
         ResponseListener responseListener = wrapResponseListener(responseConverter, listener, ignores);
         client.performRequestAsync(req, responseListener);
     }
 
-    private static void addHeaders(Request request, Header... headers) {
-        Objects.requireNonNull(headers, "headers cannot be null");
-        for (Header header : headers) {
-            request.addHeader(header.getName(), header.getValue());
-        }
-    }
-
     final <Resp> ResponseListener wrapResponseListener(CheckedFunction<Response, Resp, IOException> responseConverter,
                                                         ActionListener<Resp> actionListener, Set<Integer> ignores) {
         return new ResponseListener() {
@@ -746,6 +799,15 @@ public class RestHighLevelClient implements Closeable {
         }
     }
 
+    private static RequestOptions optionsForHeaders(Header[] headers) {
+        RequestOptions.Builder options = RequestOptions.DEFAULT.toBuilder();
+        for (Header header : headers) {
+            Objects.requireNonNull(header, "header cannot be null");
+            options.addHeader(header.getName(), header.getValue());
+        }
+        return options.build();
+    }
+
     static boolean convertExistsResponse(Response response) {
         return response.getStatusLine().getStatusCode() == 200;
     }

client/rest-high-level/src/test/java/org/elasticsearch/client/CustomRestHighLevelClientTests.java

@@ -26,7 +26,6 @@ import org.apache.http.RequestLine;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.entity.ByteArrayEntity;
 import org.apache.http.entity.ContentType;
-import org.apache.http.message.BasicHeader;
 import org.apache.http.message.BasicRequestLine;
 import org.apache.http.message.BasicStatusLine;
 import org.apache.lucene.util.BytesRef;
@@ -48,11 +47,13 @@ import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
 import java.util.stream.Collectors;
 
 import static java.util.Collections.emptySet;
-import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.hasSize;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
@@ -73,12 +74,12 @@ public class CustomRestHighLevelClientTests extends ESTestCase {
             final RestClient restClient = mock(RestClient.class);
             restHighLevelClient = new CustomRestClient(restClient);
 
-            doAnswer(inv -> mockPerformRequest(((Request) inv.getArguments()[0]).getHeaders().iterator().next()))
+            doAnswer(inv -> mockPerformRequest((Request) inv.getArguments()[0]))
                     .when(restClient)
                     .performRequest(any(Request.class));
 
             doAnswer(inv -> mockPerformRequestAsync(
-                        ((Request) inv.getArguments()[0]).getHeaders().iterator().next(),
+                        ((Request) inv.getArguments()[0]),
                         (ResponseListener) inv.getArguments()[1]))
                     .when(restClient)
                     .performRequestAsync(any(Request.class), any(ResponseListener.class));
@@ -87,26 +88,32 @@ public class CustomRestHighLevelClientTests extends ESTestCase {
 
     public void testCustomEndpoint() throws IOException {
         final MainRequest request = new MainRequest();
-        final Header header = new BasicHeader("node_name", randomAlphaOfLengthBetween(1, 10));
+        String nodeName = randomAlphaOfLengthBetween(1, 10);
 
-        MainResponse response = restHighLevelClient.custom(request, header);
-        assertEquals(header.getValue(), response.getNodeName());
+        MainResponse response = restHighLevelClient.custom(request, optionsForNodeName(nodeName));
+        assertEquals(nodeName, response.getNodeName());
 
-        response = restHighLevelClient.customAndParse(request, header);
-        assertEquals(header.getValue(), response.getNodeName());
+        response = restHighLevelClient.customAndParse(request, optionsForNodeName(nodeName));
+        assertEquals(nodeName, response.getNodeName());
     }
 
     public void testCustomEndpointAsync() throws Exception {
         final MainRequest request = new MainRequest();
-        final Header header = new BasicHeader("node_name", randomAlphaOfLengthBetween(1, 10));
+        String nodeName = randomAlphaOfLengthBetween(1, 10);
 
         PlainActionFuture<MainResponse> future = PlainActionFuture.newFuture();
-        restHighLevelClient.customAsync(request, future, header);
-        assertEquals(header.getValue(), future.get().getNodeName());
+        restHighLevelClient.customAsync(request, optionsForNodeName(nodeName), future);
+        assertEquals(nodeName, future.get().getNodeName());
 
         future = PlainActionFuture.newFuture();
-        restHighLevelClient.customAndParseAsync(request, future, header);
-        assertEquals(header.getValue(), future.get().getNodeName());
+        restHighLevelClient.customAndParseAsync(request, optionsForNodeName(nodeName), future);
+        assertEquals(nodeName, future.get().getNodeName());
+    }
+
+    private static RequestOptions optionsForNodeName(String nodeName) {
+        RequestOptions.Builder options = RequestOptions.DEFAULT.toBuilder();
+        options.addHeader("node_name", nodeName);
+        return options.build();
     }
 
     /**
@@ -115,27 +122,27 @@ public class CustomRestHighLevelClientTests extends ESTestCase {
      */
     @SuppressForbidden(reason = "We're forced to uses Class#getDeclaredMethods() here because this test checks protected methods")
     public void testMethodsVisibility() throws ClassNotFoundException {
-        final String[] methodNames = new String[]{"performRequest",
-                                                  "performRequestAsync",
+        final String[] methodNames = new String[]{"parseEntity",
+                                                  "parseResponseException",
+                                                  "performRequest",
                                                   "performRequestAndParseEntity",
-                                                  "performRequestAsyncAndParseEntity",
-                                                  "parseEntity",
-                                                  "parseResponseException"};
+                                                  "performRequestAsync",
+                                                  "performRequestAsyncAndParseEntity"};
 
-        final List<String> protectedMethods =  Arrays.stream(RestHighLevelClient.class.getDeclaredMethods())
+        final Set<String> protectedMethods =  Arrays.stream(RestHighLevelClient.class.getDeclaredMethods())
                                                      .filter(method -> Modifier.isProtected(method.getModifiers()))
                                                      .map(Method::getName)
-                                                     .collect(Collectors.toList());
+                                                     .collect(Collectors.toCollection(TreeSet::new));
 
-        assertThat(protectedMethods, containsInAnyOrder(methodNames));
+        assertThat(protectedMethods, contains(methodNames));
     }
 
     /**
-     * Mocks the asynchronous request execution by calling the {@link #mockPerformRequest(Header)} method.
+     * Mocks the asynchronous request execution by calling the {@link #mockPerformRequest(Request)} method.
      */
-    private Void mockPerformRequestAsync(Header httpHeader, ResponseListener responseListener) {
+    private Void mockPerformRequestAsync(Request request, ResponseListener responseListener) {
         try {
-            responseListener.onSuccess(mockPerformRequest(httpHeader));
+            responseListener.onSuccess(mockPerformRequest(request));
         } catch (IOException e) {
             responseListener.onFailure(e);
         }
@@ -145,7 +152,9 @@ public class CustomRestHighLevelClientTests extends ESTestCase {
     /**
      * Mocks the synchronous request execution like if it was executed by Elasticsearch.
      */
-    private Response mockPerformRequest(Header httpHeader) throws IOException {
+    private Response mockPerformRequest(Request request) throws IOException {
+        assertThat(request.getOptions().getHeaders(), hasSize(1));
+        Header httpHeader = request.getOptions().getHeaders().get(0);
         final Response mockResponse = mock(Response.class);
         when(mockResponse.getHost()).thenReturn(new HttpHost("localhost", 9200));
 
@@ -171,20 +180,20 @@ public class CustomRestHighLevelClientTests extends ESTestCase {
             super(restClient, RestClient::close, Collections.emptyList());
         }
 
-        MainResponse custom(MainRequest mainRequest, Header... headers) throws IOException {
-            return performRequest(mainRequest, this::toRequest, this::toResponse, emptySet(), headers);
+        MainResponse custom(MainRequest mainRequest, RequestOptions options) throws IOException {
+            return performRequest(mainRequest, this::toRequest, options, this::toResponse, emptySet());
         }
 
-        MainResponse customAndParse(MainRequest mainRequest, Header... headers) throws IOException {
-            return performRequestAndParseEntity(mainRequest, this::toRequest, MainResponse::fromXContent, emptySet(), headers);
+        MainResponse customAndParse(MainRequest mainRequest, RequestOptions options) throws IOException {
+            return performRequestAndParseEntity(mainRequest, this::toRequest, options, MainResponse::fromXContent, emptySet());
         }
 
-        void customAsync(MainRequest mainRequest, ActionListener<MainResponse> listener, Header... headers) {
-            performRequestAsync(mainRequest, this::toRequest, this::toResponse, listener, emptySet(), headers);
+        void customAsync(MainRequest mainRequest, RequestOptions options, ActionListener<MainResponse> listener) {
+            performRequestAsync(mainRequest, this::toRequest, options, this::toResponse, listener, emptySet());
         }
 
-        void customAndParseAsync(MainRequest mainRequest, ActionListener<MainResponse> listener, Header... headers) {
-            performRequestAsyncAndParseEntity(mainRequest, this::toRequest, MainResponse::fromXContent, listener, emptySet(), headers);
+        void customAndParseAsync(MainRequest mainRequest, RequestOptions options, ActionListener<MainResponse> listener) {
+            performRequestAsyncAndParseEntity(mainRequest, this::toRequest, options, MainResponse::fromXContent, listener, emptySet());
         }
 
         Request toRequest(MainRequest mainRequest) throws IOException {

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

@@ -19,17 +19,11 @@
 
 package org.elasticsearch.client;
 
-import org.apache.http.Header;
 import org.apache.http.HttpEntity;
 import org.apache.http.entity.ContentType;
-import org.apache.http.message.BasicHeader;
 import org.apache.http.nio.entity.NStringEntity;
-import org.apache.http.nio.protocol.HttpAsyncResponseConsumer;
 
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
@@ -42,11 +36,9 @@ public final class Request {
     private final String method;
     private final String endpoint;
     private final Map<String, String> parameters = new HashMap<>();
-    private final List<Header> headers = new ArrayList<>();
 
     private HttpEntity entity;
-    private HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory =
-            HttpAsyncResponseConsumerFactory.DEFAULT;
+    private RequestOptions options = RequestOptions.DEFAULT;
 
     /**
      * Create the {@linkplain Request}.
@@ -127,40 +119,29 @@ public final class Request {
     }
 
     /**
-     * Add the provided header to the request.
+     * Set the portion of an HTTP request to Elasticsearch that can be
+     * manipulated without changing Elasticsearch's behavior.
      */
-    public void addHeader(String name, String value) {
-        Objects.requireNonNull(name, "header name cannot be null");
-        Objects.requireNonNull(value, "header value cannot be null");
-        this.headers.add(new ReqHeader(name, value));
+    public void setOptions(RequestOptions options) {
+        Objects.requireNonNull(options, "options cannot be null");
+        this.options = options;
     }
 
     /**
-     * Headers to attach to the request.
+     * Set the portion of an HTTP request to Elasticsearch that can be
+     * manipulated without changing Elasticsearch's behavior.
      */
-    List<Header> getHeaders() {
-        return Collections.unmodifiableList(headers);
+    public void setOptions(RequestOptions.Builder options) {
+        Objects.requireNonNull(options, "options cannot be null");
+        this.options = options.build();
     }
 
     /**
-     * set the {@link HttpAsyncResponseConsumerFactory} used to create one
-     * {@link HttpAsyncResponseConsumer} callback per retry. Controls how the
-     * response body gets streamed from a non-blocking HTTP connection on the
-     * client side.
+     * Get the portion of an HTTP request to Elasticsearch that can be
+     * manipulated without changing Elasticsearch's behavior.
      */
-    public void setHttpAsyncResponseConsumerFactory(HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory) {
-        this.httpAsyncResponseConsumerFactory =
-                Objects.requireNonNull(httpAsyncResponseConsumerFactory, "httpAsyncResponseConsumerFactory cannot be null");
-    }
-
-    /**
-     * The {@link HttpAsyncResponseConsumerFactory} used to create one
-     * {@link HttpAsyncResponseConsumer} callback per retry. Controls how the
-     * response body gets streamed from a non-blocking HTTP connection on the
-     * client side.
-     */
-    public HttpAsyncResponseConsumerFactory getHttpAsyncResponseConsumerFactory() {
-        return httpAsyncResponseConsumerFactory;
+    public RequestOptions getOptions() {
+        return options;
     }
 
     @Override
@@ -175,18 +156,7 @@ public final class Request {
         if (entity != null) {
             b.append(", entity=").append(entity);
         }
-        if (headers.size() > 0) {
-            b.append(", headers=");
-            for (int h = 0; h < headers.size(); h++) {
-                if (h != 0) {
-                    b.append(',');
-                }
-                b.append(headers.get(h).toString());
-            }
-        }
-        if (httpAsyncResponseConsumerFactory != HttpAsyncResponseConsumerFactory.DEFAULT) {
-            b.append(", consumerFactory=").append(httpAsyncResponseConsumerFactory);
-        }
+        b.append(", options=").append(options);
         return b.append('}').toString();
     }
 
@@ -204,40 +174,11 @@ public final class Request {
                 && endpoint.equals(other.endpoint)
                 && parameters.equals(other.parameters)
                 && Objects.equals(entity, other.entity)
-                && headers.equals(other.headers)
-                && httpAsyncResponseConsumerFactory.equals(other.httpAsyncResponseConsumerFactory);
+                && options.equals(other.options);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(method, endpoint, parameters, entity, headers.hashCode(), httpAsyncResponseConsumerFactory);
-    }
-
-    /**
-     * Custom implementation of {@link BasicHeader} that overrides equals and hashCode.
-     */
-    static final class ReqHeader extends BasicHeader {
-
-        ReqHeader(String name, String value) {
-            super(name, value);
-        }
-
-        @Override
-        public boolean equals(Object other) {
-            if (this == other) {
-                return true;
-            }
-            if (other instanceof ReqHeader) {
-                Header otherHeader = (Header) other;
-                return Objects.equals(getName(), otherHeader.getName()) &&
-                        Objects.equals(getValue(), otherHeader.getValue());
-            }
-            return false;
-        }
-
-        @Override
-        public int hashCode() {
-            return Objects.hash(getName(), getValue());
-        }
+        return Objects.hash(method, endpoint, parameters, entity, options);
     }
 }

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

@@ -0,0 +1,175 @@
+/*
+ * 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.message.BasicHeader;
+import org.apache.http.Header;
+import org.apache.http.nio.protocol.HttpAsyncResponseConsumer;
+import org.elasticsearch.client.HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+
+import java.util.ArrayList;
+
+/**
+ * The portion of an HTTP request to Elasticsearch that can be
+ * manipulated without changing Elasticsearch's behavior.
+ */
+public final class RequestOptions {
+    public static final RequestOptions DEFAULT = new Builder(
+        Collections.<Header>emptyList(), HeapBufferedResponseConsumerFactory.DEFAULT).build();
+
+    private final List<Header> headers;
+    private final HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory;
+
+    private RequestOptions(Builder builder) {
+        this.headers = Collections.unmodifiableList(new ArrayList<>(builder.headers));
+        this.httpAsyncResponseConsumerFactory = builder.httpAsyncResponseConsumerFactory;
+    }
+
+    public Builder toBuilder() {
+        Builder builder = new Builder(headers, httpAsyncResponseConsumerFactory);
+        return builder;
+    }
+
+    /**
+     * Headers to attach to the request.
+     */
+    public List<Header> getHeaders() {
+        return headers;
+    }
+
+    /**
+     * The {@link HttpAsyncResponseConsumerFactory} used to create one
+     * {@link HttpAsyncResponseConsumer} callback per retry. Controls how the
+     * response body gets streamed from a non-blocking HTTP connection on the
+     * client side.
+     */
+    public HttpAsyncResponseConsumerFactory getHttpAsyncResponseConsumerFactory() {
+        return httpAsyncResponseConsumerFactory;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder b = new StringBuilder();
+        b.append("RequestOptions{");
+        if (headers.size() > 0) {
+            b.append(", headers=");
+            for (int h = 0; h < headers.size(); h++) {
+                if (h != 0) {
+                    b.append(',');
+                }
+                b.append(headers.get(h).toString());
+            }
+        }
+        if (httpAsyncResponseConsumerFactory != HttpAsyncResponseConsumerFactory.DEFAULT) {
+            b.append(", consumerFactory=").append(httpAsyncResponseConsumerFactory);
+        }
+        return b.append('}').toString();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null || (obj.getClass() != getClass())) {
+            return false;
+        }
+        if (obj == this) {
+            return true;
+        }
+
+        RequestOptions other = (RequestOptions) obj;
+        return headers.equals(other.headers)
+                && httpAsyncResponseConsumerFactory.equals(other.httpAsyncResponseConsumerFactory);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(headers, httpAsyncResponseConsumerFactory);
+    }
+
+    public static class Builder {
+        private final List<Header> headers;
+        private HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory;
+
+        private Builder(List<Header> headers, HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory) {
+            this.headers = new ArrayList<>(headers);
+            this.httpAsyncResponseConsumerFactory = httpAsyncResponseConsumerFactory;
+        }
+
+        /**
+         * Build the {@linkplain RequestOptions}.
+         */
+        public RequestOptions build() {
+            return new RequestOptions(this);
+        }
+
+        /**
+         * Add the provided header to the request.
+         */
+        public void addHeader(String name, String value) {
+            Objects.requireNonNull(name, "header name cannot be null");
+            Objects.requireNonNull(value, "header value cannot be null");
+            this.headers.add(new ReqHeader(name, value));
+        }
+
+        /**
+         * set the {@link HttpAsyncResponseConsumerFactory} used to create one
+         * {@link HttpAsyncResponseConsumer} callback per retry. Controls how the
+         * response body gets streamed from a non-blocking HTTP connection on the
+         * client side.
+         */
+        public void setHttpAsyncResponseConsumerFactory(HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory) {
+            this.httpAsyncResponseConsumerFactory =
+                    Objects.requireNonNull(httpAsyncResponseConsumerFactory, "httpAsyncResponseConsumerFactory cannot be null");
+        }
+    }
+
+    /**
+     * Custom implementation of {@link BasicHeader} that overrides equals and
+     * hashCode so it is easier to test equality of {@link RequestOptions}.
+     */
+    static final class ReqHeader extends BasicHeader {
+
+        ReqHeader(String name, String value) {
+            super(name, value);
+        }
+
+        @Override
+        public boolean equals(Object other) {
+            if (this == other) {
+                return true;
+            }
+            if (other instanceof ReqHeader) {
+                Header otherHeader = (Header) other;
+                return Objects.equals(getName(), otherHeader.getName()) &&
+                        Objects.equals(getValue(), otherHeader.getValue());
+            }
+            return false;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(getName(), getValue());
+        }
+    }
+}

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

@@ -312,8 +312,7 @@ public class RestClient implements Closeable {
         Request request = new Request(method, endpoint);
         addParameters(request, params);
         request.setEntity(entity);
-        request.setHttpAsyncResponseConsumerFactory(httpAsyncResponseConsumerFactory);
-        addHeaders(request, headers);
+        setOptions(request, httpAsyncResponseConsumerFactory, headers);
         return performRequest(request);
     }
 
@@ -427,8 +426,7 @@ public class RestClient implements Closeable {
             request = new Request(method, endpoint);
             addParameters(request, params);
             request.setEntity(entity);
-            request.setHttpAsyncResponseConsumerFactory(httpAsyncResponseConsumerFactory);
-            addHeaders(request, headers);
+            setOptions(request, httpAsyncResponseConsumerFactory, headers);
         } catch (Exception e) {
             responseListener.onFailure(e);
             return;
@@ -465,11 +463,11 @@ public class RestClient implements Closeable {
         }
         URI uri = buildUri(pathPrefix, request.getEndpoint(), requestParams);
         HttpRequestBase httpRequest = createHttpRequest(request.getMethod(), uri, request.getEntity());
-        setHeaders(httpRequest, request.getHeaders());
+        setHeaders(httpRequest, request.getOptions().getHeaders());
         FailureTrackingResponseListener failureTrackingResponseListener = new FailureTrackingResponseListener(listener);
         long startTime = System.nanoTime();
         performRequestAsync(startTime, nextHost(), httpRequest, ignoreErrorCodes,
-                request.getHttpAsyncResponseConsumerFactory(), failureTrackingResponseListener);
+                request.getOptions().getHttpAsyncResponseConsumerFactory(), failureTrackingResponseListener);
     }
 
     private void performRequestAsync(final long startTime, final HostTuple<Iterator<HttpHost>> hostTuple, final HttpRequestBase request,
@@ -891,11 +889,24 @@ public class RestClient implements Closeable {
      */
     @Deprecated
     private static void addHeaders(Request request, Header... headers) {
+        setOptions(request, RequestOptions.DEFAULT.getHttpAsyncResponseConsumerFactory(), headers);
+    }
+
+    /**
+     * Add all headers from the provided varargs argument to a {@link Request}. This only exists
+     * to support methods that exist for backwards compatibility.
+     */
+    @Deprecated
+    private static void setOptions(Request request, HttpAsyncResponseConsumerFactory httpAsyncResponseConsumerFactory,
+            Header... headers) {
         Objects.requireNonNull(headers, "headers cannot be null");
+        RequestOptions.Builder options = request.getOptions().toBuilder();
         for (Header header : headers) {
             Objects.requireNonNull(header, "header cannot be null");
-            request.addHeader(header.getName(), header.getValue());
+            options.addHeader(header.getName(), header.getValue());
         }
+        options.setHttpAsyncResponseConsumerFactory(httpAsyncResponseConsumerFactory);
+        request.setOptions(options);
     }
 
     /**

client/rest/src/test/java/org/elasticsearch/client/RequestOptionsTests.java

@@ -0,0 +1,142 @@
+/*
+ * 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.Header;
+import org.elasticsearch.client.HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+public class RequestOptionsTests extends RestClientTestCase {
+    public void testDefault() {
+        assertEquals(Collections.<Header>emptyList(), RequestOptions.DEFAULT.getHeaders());
+        assertEquals(HttpAsyncResponseConsumerFactory.DEFAULT, RequestOptions.DEFAULT.getHttpAsyncResponseConsumerFactory());
+        assertEquals(RequestOptions.DEFAULT, RequestOptions.DEFAULT.toBuilder().build());
+    }
+
+    public void testAddHeader() {
+        try {
+            randomBuilder().addHeader(null, randomAsciiLettersOfLengthBetween(3, 10));
+            fail("expected failure");
+        } catch (NullPointerException e) {
+            assertEquals("header name cannot be null", e.getMessage());
+        }
+
+        try {
+            randomBuilder().addHeader(randomAsciiLettersOfLengthBetween(3, 10), null);
+            fail("expected failure");
+        } catch (NullPointerException e) {
+            assertEquals("header value cannot be null", e.getMessage());
+        }
+
+        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
+        int numHeaders = between(0, 5);
+        List<Header> headers = new ArrayList<>();
+        for (int i = 0; i < numHeaders; i++) {
+            Header header = new RequestOptions.ReqHeader(randomAsciiAlphanumOfLengthBetween(5, 10), randomAsciiAlphanumOfLength(3));
+            headers.add(header);
+            builder.addHeader(header.getName(), header.getValue());
+        }
+        RequestOptions options = builder.build();
+        assertEquals(headers, options.getHeaders());
+
+        try {
+            options.getHeaders().add(
+                    new RequestOptions.ReqHeader(randomAsciiAlphanumOfLengthBetween(5, 10), randomAsciiAlphanumOfLength(3)));
+            fail("expected failure");
+        } catch (UnsupportedOperationException e) {
+            assertNull(e.getMessage());
+        }
+    }
+
+    public void testSetHttpAsyncResponseConsumerFactory() {
+        try {
+            RequestOptions.DEFAULT.toBuilder().setHttpAsyncResponseConsumerFactory(null);
+            fail("expected failure");
+        } catch (NullPointerException e) {
+            assertEquals("httpAsyncResponseConsumerFactory cannot be null", e.getMessage());
+        }
+
+        HttpAsyncResponseConsumerFactory factory = mock(HttpAsyncResponseConsumerFactory.class);
+        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
+        builder.setHttpAsyncResponseConsumerFactory(factory);
+        RequestOptions options = builder.build();
+        assertSame(factory, options.getHttpAsyncResponseConsumerFactory());
+    }
+
+    public void testEqualsAndHashCode() {
+        RequestOptions request = randomBuilder().build();
+        assertEquals(request, request);
+
+        RequestOptions copy = copy(request);
+        assertEquals(request, copy);
+        assertEquals(copy, request);
+        assertEquals(request.hashCode(), copy.hashCode());
+
+        RequestOptions mutant = mutate(request);
+        assertNotEquals(request, mutant);
+        assertNotEquals(mutant, request);
+    }
+
+    static RequestOptions.Builder randomBuilder() {
+        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
+
+        if (randomBoolean()) {
+            int headerCount = between(1, 5);
+            for (int i = 0; i < headerCount; i++) {
+                builder.addHeader(randomAsciiAlphanumOfLength(3), randomAsciiAlphanumOfLength(3));
+            }
+        }
+
+        if (randomBoolean()) {
+            builder.setHttpAsyncResponseConsumerFactory(new HeapBufferedResponseConsumerFactory(1));
+        }
+
+        return builder;
+    }
+
+    private static RequestOptions copy(RequestOptions options) {
+        return options.toBuilder().build();
+    }
+
+    private static RequestOptions mutate(RequestOptions options) {
+        RequestOptions.Builder mutant = options.toBuilder();
+        int mutationType = between(0, 1);
+        switch (mutationType) {
+        case 0:
+            mutant.addHeader("extra", "m");
+            return mutant.build();
+        case 1:
+            mutant.setHttpAsyncResponseConsumerFactory(new HeapBufferedResponseConsumerFactory(5));
+            return mutant.build();
+        default:
+            throw new UnsupportedOperationException("Unknown mutation type [" + mutationType + "]");
+        }
+    }
+}

client/rest/src/test/java/org/elasticsearch/client/RequestTests.java

@@ -37,6 +37,7 @@ import java.util.Map;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.fail;
 
 public class RequestTests extends RestClientTestCase {
@@ -127,33 +128,33 @@ public class RequestTests extends RestClientTestCase {
         assertEquals(json, new String(os.toByteArray(), ContentType.APPLICATION_JSON.getCharset()));
     }
 
-    public void testAddHeader() {
+    public void testSetOptions() {
         final String method = randomFrom(new String[] {"GET", "PUT", "POST", "HEAD", "DELETE"});
         final String endpoint = randomAsciiLettersOfLengthBetween(1, 10);
         Request request = new Request(method, endpoint);
 
         try {
-            request.addHeader(null, randomAsciiLettersOfLengthBetween(3, 10));
+            request.setOptions((RequestOptions) null);
             fail("expected failure");
         } catch (NullPointerException e) {
-            assertEquals("header name cannot be null", e.getMessage());
+            assertEquals("options cannot be null", e.getMessage());
         }
 
         try {
-            request.addHeader(randomAsciiLettersOfLengthBetween(3, 10), null);
+            request.setOptions((RequestOptions.Builder) null);
             fail("expected failure");
         } catch (NullPointerException e) {
-            assertEquals("header value cannot be null", e.getMessage());
+            assertEquals("options cannot be null", e.getMessage());
         }
 
-        int numHeaders = between(0, 5);
-        List<Header> headers = new ArrayList<>();
-        for (int i = 0; i < numHeaders; i++) {
-            Header header = new Request.ReqHeader(randomAsciiAlphanumOfLengthBetween(5, 10), randomAsciiAlphanumOfLength(3));
-            headers.add(header);
-            request.addHeader(header.getName(), header.getValue());
-        }
-        assertEquals(headers, new ArrayList<>(request.getHeaders()));
+        RequestOptions.Builder builder = RequestOptionsTests.randomBuilder();
+        request.setOptions(builder);
+        assertEquals(builder.build(), request.getOptions());
+
+        builder = RequestOptionsTests.randomBuilder();
+        RequestOptions options = builder.build();
+        request.setOptions(options);
+        assertSame(options, request.getOptions());
     }
 
     public void testEqualsAndHashCode() {
@@ -193,14 +194,9 @@ public class RequestTests extends RestClientTestCase {
         }
 
         if (randomBoolean()) {
-            int headerCount = between(1, 5);
-            for (int i = 0; i < headerCount; i++) {
-                request.addHeader(randomAsciiAlphanumOfLength(3), randomAsciiAlphanumOfLength(3));
-            }
-        }
-
-        if (randomBoolean()) {
-            request.setHttpAsyncResponseConsumerFactory(new HeapBufferedResponseConsumerFactory(1));
+            RequestOptions.Builder options = request.getOptions().toBuilder();
+            options.setHttpAsyncResponseConsumerFactory(new HeapBufferedResponseConsumerFactory(1));
+            request.setOptions(options);
         }
 
         return request;
@@ -222,7 +218,7 @@ public class RequestTests extends RestClientTestCase {
             return mutant;
         }
         Request mutant = copy(request);
-        int mutationType = between(0, 3);
+        int mutationType = between(0, 2);
         switch (mutationType) {
         case 0:
             mutant.addParameter(randomAsciiAlphanumOfLength(mutant.getParameters().size() + 4), "extra");
@@ -231,10 +227,9 @@ public class RequestTests extends RestClientTestCase {
             mutant.setJsonEntity("mutant"); // randomRequest can't produce this value
             return mutant;
         case 2:
-            mutant.addHeader("extra", "m");
-            return mutant;
-        case 3:
-            mutant.setHttpAsyncResponseConsumerFactory(new HeapBufferedResponseConsumerFactory(5));
+            RequestOptions.Builder options = mutant.getOptions().toBuilder();
+            options.addHeader("extra", "m");
+            mutant.setOptions(options);
             return mutant;
         default:
             throw new UnsupportedOperationException("Unknown mutation type [" + mutationType + "]");
@@ -246,9 +241,6 @@ public class RequestTests extends RestClientTestCase {
             to.addParameter(param.getKey(), param.getValue());
         }
         to.setEntity(from.getEntity());
-        for (Header header : from.getHeaders()) {
-            to.addHeader(header.getName(), header.getValue());
-        }
-        to.setHttpAsyncResponseConsumerFactory(from.getHttpAsyncResponseConsumerFactory());
+        to.setOptions(from.getOptions());
     }
 }

client/rest/src/test/java/org/elasticsearch/client/RestClientSingleHostIntegTests.java

@@ -378,9 +378,11 @@ public class RestClientSingleHostIntegTests extends RestClientTestCase {
         String requestBody = "{ \"field\": \"value\" }";
         Request request = new Request(method, "/" + statusCode);
         request.setJsonEntity(requestBody);
+        RequestOptions.Builder options = request.getOptions().toBuilder();
         for (Header header : headers) {
-            request.addHeader(header.getName(), header.getValue());
+            options.addHeader(header.getName(), header.getValue());
         }
+        request.setOptions(options);
         Response esResponse;
         try {
             esResponse = restClient.performRequest(request);

client/rest/src/test/java/org/elasticsearch/client/RestClientSingleHostTests.java

@@ -362,9 +362,11 @@ public class RestClientSingleHostTests extends RestClientTestCase {
             final Header[] requestHeaders = RestClientTestUtil.randomHeaders(getRandom(), "Header");
             final int statusCode = randomStatusCode(getRandom());
             Request request = new Request(method, "/" + statusCode);
+            RequestOptions.Builder options = request.getOptions().toBuilder();
             for (Header requestHeader : requestHeaders) {
-                request.addHeader(requestHeader.getName(), requestHeader.getValue());
+                options.addHeader(requestHeader.getName(), requestHeader.getValue());
             }
+            request.setOptions(options);
             Response esResponse;
             try {
                 esResponse = restClient.performRequest(request);
@@ -438,11 +440,13 @@ public class RestClientSingleHostTests extends RestClientTestCase {
         final Set<String> uniqueNames = new HashSet<>();
         if (randomBoolean()) {
             Header[] headers = RestClientTestUtil.randomHeaders(getRandom(), "Header");
+            RequestOptions.Builder options = request.getOptions().toBuilder();
             for (Header header : headers) {
-                request.addHeader(header.getName(), header.getValue());
-                expectedRequest.addHeader(new Request.ReqHeader(header.getName(), header.getValue()));
+                options.addHeader(header.getName(), header.getValue());
+                expectedRequest.addHeader(new RequestOptions.ReqHeader(header.getName(), header.getValue()));
                 uniqueNames.add(header.getName());
             }
+            request.setOptions(options);
         }
         for (Header defaultHeader : defaultHeaders) {
             // request level headers override default headers

client/rest/src/test/java/org/elasticsearch/client/documentation/RestClientDocumentation.java

@@ -38,6 +38,7 @@ import org.apache.http.ssl.SSLContexts;
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseListener;
 import org.elasticsearch.client.RestClient;
@@ -171,14 +172,22 @@ public class RestClientDocumentation {
             //tag::rest-client-body-shorter
             request.setJsonEntity("{\"json\":\"text\"}");
             //end::rest-client-body-shorter
-            //tag::rest-client-headers
-            request.addHeader("Accept", "text/plain");
-            request.addHeader("Cache-Control", "no-cache");
-            //end::rest-client-headers
-            //tag::rest-client-response-consumer
-            request.setHttpAsyncResponseConsumerFactory(
-                    new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
-            //end::rest-client-response-consumer
+            {
+                //tag::rest-client-headers
+                RequestOptions.Builder options = request.getOptions().toBuilder();
+                options.addHeader("Accept", "text/plain");
+                options.addHeader("Cache-Control", "no-cache");
+                request.setOptions(options);
+                //end::rest-client-headers
+            }
+            {
+                //tag::rest-client-response-consumer
+                RequestOptions.Builder options = request.getOptions().toBuilder();
+                options.setHttpAsyncResponseConsumerFactory(
+                        new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
+                request.setOptions(options);
+                //end::rest-client-response-consumer
+            }
         }
         {
             HttpEntity[] documents = new HttpEntity[10];

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

@@ -31,6 +31,7 @@ import org.elasticsearch.action.support.ActionFilter;
 import org.elasticsearch.action.termvectors.MultiTermVectorsRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
@@ -221,8 +222,10 @@ public class ContextAndHeaderTransportIT extends HttpSmokeTestCase {
     public void testThatRelevantHttpHeadersBecomeRequestHeaders() throws IOException {
         final String IRRELEVANT_HEADER = "SomeIrrelevantHeader";
         Request request = new Request("GET", "/" + queryIndex + "/_search");
-        request.addHeader(CUSTOM_HEADER, randomHeaderValue);
-        request.addHeader(IRRELEVANT_HEADER, randomHeaderValue);
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader(CUSTOM_HEADER, randomHeaderValue);
+        options.addHeader(IRRELEVANT_HEADER, randomHeaderValue);
+        request.setOptions(options);
         Response response = getRestClient().performRequest(request);
         assertThat(response.getStatusLine().getStatusCode(), equalTo(200));
         List<RequestAndHeaders> searchRequests = getRequests(SearchRequest.class);

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.http;
 
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 
 import java.io.IOException;
@@ -32,8 +33,10 @@ public class CorsNotSetIT extends HttpSmokeTestCase {
     public void testCorsSettingDefaultBehaviourDoesNotReturnAnything() throws IOException {
         String corsValue = "http://localhost:9200";
         Request request = new Request("GET", "/");
-        request.addHeader("User-Agent", "Mozilla Bar");
-        request.addHeader("Origin", corsValue);
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("User-Agent", "Mozilla Bar");
+        options.addHeader("Origin", corsValue);
+        request.setOptions(options);
         Response response = getRestClient().performRequest(request);
         assertThat(response.getStatusLine().getStatusCode(), is(200));
         assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue());

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

@@ -19,6 +19,7 @@
 package org.elasticsearch.http;
 
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.settings.Settings;
@@ -55,16 +56,20 @@ public class CorsRegexIT extends HttpSmokeTestCase {
         {
             String corsValue = "http://localhost:9200";
             Request request = new Request("GET", "/");
-            request.addHeader("User-Agent", "Mozilla Bar");
-            request.addHeader("Origin", corsValue);
+            RequestOptions.Builder options = request.getOptions().toBuilder();
+            options.addHeader("User-Agent", "Mozilla Bar");
+            options.addHeader("Origin", corsValue);
+            request.setOptions(options);
             Response response = getRestClient().performRequest(request);
             assertResponseWithOriginHeader(response, corsValue);
         }
         {
             String corsValue = "https://localhost:9201";
             Request request = new Request("GET", "/");
-            request.addHeader("User-Agent", "Mozilla Bar");
-            request.addHeader("Origin", corsValue);
+            RequestOptions.Builder options = request.getOptions().toBuilder();
+            options.addHeader("User-Agent", "Mozilla Bar");
+            options.addHeader("Origin", corsValue);
+            request.setOptions(options);
             Response response = getRestClient().performRequest(request);
             assertResponseWithOriginHeader(response, corsValue);
             assertThat(response.getHeader("Access-Control-Allow-Credentials"), is("true"));
@@ -73,8 +78,10 @@ public class CorsRegexIT extends HttpSmokeTestCase {
 
     public void testThatRegularExpressionReturnsForbiddenOnNonMatch() throws IOException {
         Request request = new Request("GET", "/");
-        request.addHeader("User-Agent", "Mozilla Bar");
-        request.addHeader("Origin", "http://evil-host:9200");
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("User-Agent", "Mozilla Bar");
+        options.addHeader("Origin", "http://evil-host:9200");
+        request.setOptions(options);
         try {
             getRestClient().performRequest(request);
             fail("request should have failed");
@@ -88,7 +95,9 @@ public class CorsRegexIT extends HttpSmokeTestCase {
 
     public void testThatSendingNoOriginHeaderReturnsNoAccessControlHeader() throws IOException {
         Request request = new Request("GET", "/");
-        request.addHeader("User-Agent", "Mozilla Bar");
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("User-Agent", "Mozilla Bar");
+        request.setOptions(options);
         Response response = getRestClient().performRequest(request);
         assertThat(response.getStatusLine().getStatusCode(), is(200));
         assertThat(response.getHeader("Access-Control-Allow-Origin"), nullValue());
@@ -103,9 +112,11 @@ public class CorsRegexIT extends HttpSmokeTestCase {
     public void testThatPreFlightRequestWorksOnMatch() throws IOException {
         String corsValue = "http://localhost:9200";
         Request request = new Request("OPTIONS", "/");
-        request.addHeader("User-Agent", "Mozilla Bar");
-        request.addHeader("Origin", corsValue);
-        request.addHeader("Access-Control-Request-Method", "GET");
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("User-Agent", "Mozilla Bar");
+        options.addHeader("Origin", corsValue);
+        options.addHeader("Access-Control-Request-Method", "GET");
+        request.setOptions(options);
         Response response = getRestClient().performRequest(request);
         assertResponseWithOriginHeader(response, corsValue);
         assertNotNull(response.getHeader("Access-Control-Allow-Methods"));
@@ -114,9 +125,11 @@ public class CorsRegexIT extends HttpSmokeTestCase {
     public void testThatPreFlightRequestReturnsNullOnNonMatch() throws IOException {
         String corsValue = "http://evil-host:9200";
         Request request = new Request("OPTIONS", "/");
-        request.addHeader("User-Agent", "Mozilla Bar");
-        request.addHeader("Origin", corsValue);
-        request.addHeader("Access-Control-Request-Method", "GET");
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("User-Agent", "Mozilla Bar");
+        options.addHeader("Origin", corsValue);
+        options.addHeader("Access-Control-Request-Method", "GET");
+        request.setOptions(options);
         try {
             getRestClient().performRequest(request);
             fail("request should have failed");

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

@@ -20,6 +20,7 @@ package org.elasticsearch.http;
 
 import org.apache.http.HttpHeaders;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.test.rest.ESRestTestCase;
 
@@ -38,7 +39,9 @@ public class HttpCompressionIT extends ESRestTestCase {
 
     public void testCompressesResponseIfRequested() throws IOException {
         Request request = new Request("GET", "/");
-        request.addHeader(HttpHeaders.ACCEPT_ENCODING, GZIP_ENCODING);
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader(HttpHeaders.ACCEPT_ENCODING, GZIP_ENCODING);
+        request.setOptions(options);
         Response response = client().performRequest(request);
         assertEquals(200, response.getStatusLine().getStatusCode());
         assertEquals(GZIP_ENCODING, response.getHeader(HttpHeaders.CONTENT_ENCODING));

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

@@ -21,6 +21,7 @@ package org.elasticsearch.http;
 
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 
@@ -46,7 +47,9 @@ public class NoHandlerIT extends HttpSmokeTestCase {
     private void runTestNoHandlerRespectsAcceptHeader(
             final String accept, final String contentType, final String expect) throws IOException {
         Request request = new Request("GET", "/foo/bar/baz/qux/quux");
-        request.addHeader("Accept", accept);
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("Accept", accept);
+        request.setOptions(options);
         final ResponseException e = expectThrows(ResponseException.class,
                         () -> getRestClient().performRequest(request));
 

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

@@ -19,6 +19,7 @@
 package org.elasticsearch.http;
 
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.plugins.Plugin;
@@ -61,7 +62,9 @@ public class ResponseHeaderPluginIT extends HttpSmokeTestCase {
         }
 
         Request request = new Request("GET", "/_protected");
-        request.addHeader("Secret", "password");
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("Secret", "password");
+        request.setOptions(options);
         Response authResponse = getRestClient().performRequest(request);
         assertThat(authResponse.getStatusLine().getStatusCode(), equalTo(200));
         assertThat(authResponse.getHeader("Secret"), equalTo("granted"));

test/framework/src/main/java/org/elasticsearch/test/rest/yaml/ESClientYamlSuiteTestCase.java

@@ -322,8 +322,7 @@ public abstract class ESClientYamlSuiteTestCase extends ESRestTestCase {
         if (useDefaultNumberOfShards == false
                 && testCandidate.getTestSection().getSkipSection().getFeatures().contains("default_shards") == false) {
             final Request request = new Request("PUT", "/_template/global");
-            request.addHeader("Content-Type", XContentType.JSON.mediaTypeWithoutParameters());
-            request.setEntity(new StringEntity("{\"index_patterns\":[\"*\"],\"settings\":{\"index.number_of_shards\":2}}"));
+            request.setJsonEntity("{\"index_patterns\":[\"*\"],\"settings\":{\"index.number_of_shards\":2}}");
             adminClient().performRequest(request);
         }
 

x-pack/qa/sql/security/src/test/java/org/elasticsearch/xpack/qa/sql/security/RestSqlSecurityIT.java

@@ -9,6 +9,7 @@ import org.apache.http.HttpEntity;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.Nullable;
@@ -177,7 +178,9 @@ public class RestSqlSecurityIT extends SqlSecurityTestCase {
                 request.addParameter("mode", mode);
             }
             if (asUser != null) {
-                request.addHeader("es-security-runas-user", asUser);
+                RequestOptions.Builder options = request.getOptions().toBuilder();
+                options.addHeader("es-security-runas-user", asUser);
+                request.setOptions(options);
             }
             request.setEntity(entity);
             return toMap(client().performRequest(request));

x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/rest/RestSqlTestCase.java

@@ -10,6 +10,7 @@ import org.apache.http.HttpEntity;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.CheckedSupplier;
@@ -319,7 +320,10 @@ public abstract class RestSqlTestCase extends ESRestTestCase implements ErrorsTe
             request.addParameter("mode", mode);        // JDBC or PLAIN mode
         }
         if (randomBoolean()) {
-            request.addHeader("Accept", randomFrom("*/*", "application/json"));
+            // JSON is the default but randomly set it sometime for extra coverage
+            RequestOptions.Builder options = request.getOptions().toBuilder();
+            options.addHeader("Accept", randomFrom("*/*", "application/json"));
+            request.setOptions(options);
         }
         request.setEntity(sql);
         Response response = client().performRequest(request);
@@ -536,7 +540,9 @@ public abstract class RestSqlTestCase extends ESRestTestCase implements ErrorsTe
         Request request = new Request("POST", "/_xpack/sql" + suffix);
         request.addParameter("error_trace", "true");
         request.setEntity(entity);
-        request.addHeader("Accept", accept);
+        RequestOptions.Builder options = request.getOptions().toBuilder();
+        options.addHeader("Accept", accept);
+        request.setOptions(options);
         Response response = client().performRequest(request);
         return new Tuple<>(
                 Streams.copyToString(new InputStreamReader(response.getEntity().getContent(), StandardCharsets.UTF_8)),