Add support for rest compatibility headers to the HLRC

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

@@ -8,7 +8,9 @@
 
 package org.elasticsearch.client;
 
+import org.apache.http.Header;
 import org.apache.http.HttpEntity;
+import org.apache.http.HttpRequest;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Build;
@@ -253,11 +255,16 @@
 public class RestHighLevelClient implements Closeable {
 
     private static final Logger logger = LogManager.getLogger(RestHighLevelClient.class);
+    /**
+     * Environment variable determining whether to send the 7.x compatibility header
+     */
+    public static final String API_VERSIONING_ENV_VARIABLE = "ELASTIC_CLIENT_APIVERSIONING";
 
     // To be called using performClientRequest and performClientRequestAsync to ensure version compatibility check
     private final RestClient client;
     private final NamedXContentRegistry registry;
     private final CheckedConsumer<RestClient, IOException> doClose;
+    private final boolean useAPICompatibility;
 
     /** Do not access directly but through getVersionValidationFuture() */
     private volatile ListenableFuture<Optional<String>> versionValidationFuture;
@@ -310,11 +317,28 @@ protected RestHighLevelClient(RestClientBuilder restClientBuilder, List<NamedXCo
      */
     protected RestHighLevelClient(RestClient restClient, CheckedConsumer<RestClient, IOException> doClose,
                                   List<NamedXContentRegistry.Entry> namedXContentEntries) {
+        this(restClient, doClose, namedXContentEntries, null);
+    }
+
+    /**
+     * Creates a {@link RestHighLevelClient} given the low level {@link RestClient} that it should use to perform requests and
+     * a list of entries that allow to parse custom response sections added to Elasticsearch through plugins.
+     * This constructor can be called by subclasses in case an externally created low-level REST client needs to be provided.
+     * The consumer argument allows to control what needs to be done when the {@link #close()} method is called.
+     * Also subclasses can provide parsers for custom response sections added to Elasticsearch through plugins.
+     */
+    protected RestHighLevelClient(RestClient restClient, CheckedConsumer<RestClient, IOException> doClose,
+                                  List<NamedXContentRegistry.Entry> namedXContentEntries, Boolean useAPICompatibility) {
         this.client = Objects.requireNonNull(restClient, "restClient must not be null");
         this.doClose = Objects.requireNonNull(doClose, "doClose consumer must not be null");
         this.registry = new NamedXContentRegistry(
-                Stream.of(getDefaultNamedXContents().stream(), getProvidedNamedXContents().stream(), namedXContentEntries.stream())
-                    .flatMap(Function.identity()).collect(toList()));
+            Stream.of(getDefaultNamedXContents().stream(), getProvidedNamedXContents().stream(), namedXContentEntries.stream())
+                .flatMap(Function.identity()).collect(toList()));
+        if (useAPICompatibility == null && "true".equals(System.getenv(API_VERSIONING_ENV_VARIABLE))) {
+            this.useAPICompatibility = true;
+        } else {
+            this.useAPICompatibility = Boolean.TRUE.equals(useAPICompatibility);
+        }
     }
 
     /**
@@ -2016,7 +2040,82 @@ protected static boolean convertExistsResponse(Response response) {
         return response.getStatusLine().getStatusCode() == 200;
     }
 
+    private enum EntityType {
+        JSON() {
+            @Override
+            public String header() {
+                return "application/json";
+            }
+            @Override
+            public String compatibleHeader() {
+                return "application/vnd.elasticsearch+json; compatible-with=7";
+            }
+        },
+        NDJSON() {
+            @Override
+            public String header() {
+                return "application/x-ndjson";
+            }
+            @Override
+            public String compatibleHeader() {
+                return "application/vnd.elasticsearch+x-ndjson; compatible-with=7";
+            }
+        },
+        STAR() {
+            @Override
+            public String header() {
+                return "application/*";
+            }
+            @Override
+            public String compatibleHeader() {
+                return "application/vnd.elasticsearch+json; compatible-with=7";
+            }
+        },
+        YAML() {
+            @Override
+            public String header() {
+                return "application/yaml";
+            }
+            @Override
+            public String compatibleHeader() {
+                return "application/vnd.elasticsearch+yaml; compatible-with=7";
+            }
+        },
+        SMILE() {
+            @Override
+            public String header() {
+                return "application/smile";
+            }
+            @Override
+            public String compatibleHeader() {
+                return "application/vnd.elasticsearch+smile; compatible-with=7";
+            }
+        },
+        CBOR() {
+            @Override
+            public String header() {
+                return "application/cbor";
+            }
+            @Override
+            public String compatibleHeader() {
+                return "application/vnd.elasticsearch+cbor; compatible-with=7";
+            }
+        };
+
+        public abstract String header();
+        public abstract String compatibleHeader();
+
+        @Override
+        public String toString() {
+            return header();
+        }
+    }
+
     private Cancellable performClientRequestAsync(Request request, ResponseListener listener) {
+        // Add compatibility request headers if compatibility mode has been enabled
+        if (this.useAPICompatibility) {
+            modifyRequestForCompatibility(request);
+        }
 
         ListenableFuture<Optional<String>> versionCheck = getVersionValidationFuture();
 
@@ -2068,7 +2167,71 @@ public void onFailure(Exception e) {
         return result;
     };
 
+
+    /**
+     * Go through all the request's existing headers, looking for {@code headerName} headers and if they exist,
+     * changing them to use version compatibility. If no request headers are changed, modify the entity type header if appropriate
+     */
+    boolean addCompatibilityFor(RequestOptions.Builder newOptions,  Header entityHeader, String headerName) {
+        // Modify any existing "Content-Type" headers on the request to use the version compatibility, if available
+        boolean contentTypeModified = false;
+        for (Header header : newOptions.getHeaders()) {
+            if (headerName.equals(header.getName()) == false) {
+                continue;
+            }
+            contentTypeModified = contentTypeModified || modifyHeader(newOptions, header, headerName);
+        }
+
+        // If there were no request-specific headers, modify the request entity's header to be compatible
+        if (entityHeader != null && contentTypeModified == false) {
+            contentTypeModified = modifyHeader(newOptions, entityHeader, headerName);
+        }
+
+        return contentTypeModified;
+    }
+
+    /**
+     * Modify the given header to be version compatible, if necessary.
+     * Returns true if a modification was made, false otherwise.
+     */
+    boolean modifyHeader(RequestOptions.Builder newOptions, Header header, String headerName) {
+        for (EntityType type : EntityType.values()) {
+            final String headerValue = header.getValue();
+            if (headerValue.startsWith(type.header())) {
+                String newHeaderValue = headerValue.replace(type.header(), type.compatibleHeader());
+                newOptions.removeHeader(header.getName());
+                newOptions.addHeader(headerName, newHeaderValue);
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Make all necessary changes to support API compatibility for the given request. This includes
+     * modifying the "Content-Type" and "Accept" headers if present, or modifying the header based
+     * on the request's entity type.
+     */
+    void modifyRequestForCompatibility(Request request) {
+        final Header entityHeader = request.getEntity() == null ? null : request.getEntity().getContentType();
+        final RequestOptions.Builder newOptions = request.getOptions().toBuilder();
+
+        addCompatibilityFor(newOptions, entityHeader, "Content-Type");
+        if (request.getOptions().containsHeader("Accept")) {
+            addCompatibilityFor(newOptions, entityHeader, "Accept");
+        } else {
+            // There is no entity, and no existing accept header, but we still need one
+            // with compatibility, so use the compatible JSON (default output) format
+            newOptions.addHeader("Accept", EntityType.JSON.compatibleHeader());
+        }
+        request.setOptions(newOptions);
+    }
+
     private Response performClientRequest(Request request) throws IOException {
+        // Add compatibility request headers if compatibility mode has been enabled
+        if (this.useAPICompatibility) {
+            modifyRequestForCompatibility(request);
+        }
 
         Optional<String> versionValidation;
         try {

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

@@ -0,0 +1,54 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.client;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.core.CheckedConsumer;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Helper to build a {@link RestHighLevelClient}, allowing setting the low-level client that
+ * should be used as well as whether API compatibility should be used.
+ */
+
+public class RestHighLevelClientBuilder {
+    private final RestClient restClient;
+    private CheckedConsumer<RestClient, IOException> closeHandler = RestClient::close;
+    private List<NamedXContentRegistry.Entry> namedXContentEntries = Collections.emptyList();
+    private Boolean apiCompatibilityMode = null;
+
+    public RestHighLevelClientBuilder(RestClient restClient) {
+        this.restClient = restClient;
+    }
+
+    public RestHighLevelClientBuilder closeHandler(CheckedConsumer<RestClient, IOException> closeHandler) {
+        this.closeHandler = closeHandler;
+        return this;
+    }
+
+    public RestHighLevelClientBuilder namedXContentEntries(List<NamedXContentRegistry.Entry> namedXContentEntries) {
+        this.namedXContentEntries = namedXContentEntries;
+        return this;
+    }
+
+    public RestHighLevelClientBuilder setApiCompatibilityMode(Boolean enabled) {
+        this.apiCompatibilityMode = enabled;
+        return this;
+    }
+
+    public RestHighLevelClient build() {
+        return new RestHighLevelClient(this.restClient, this.closeHandler, this.namedXContentEntries, this.apiCompatibilityMode);
+    }
+}

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

@@ -10,6 +10,9 @@
 
 import com.fasterxml.jackson.core.JsonParseException;
 
+import joptsimple.internal.Strings;
+
+import org.apache.http.Header;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpHost;
 import org.apache.http.HttpResponse;
@@ -19,6 +22,8 @@
 import org.apache.http.StatusLine;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.message.BasicHeader;
 import org.apache.http.message.BasicHttpResponse;
 import org.apache.http.message.BasicRequestLine;
 import org.apache.http.message.BasicStatusLine;
@@ -139,6 +144,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -147,6 +153,7 @@
 import static org.hamcrest.CoreMatchers.endsWith;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.hasItems;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.argThat;
@@ -1226,6 +1233,88 @@ public void testCancellationForwarding() throws Exception {
         verify(cancellable, times(1)).cancel();
     }
 
+    public void testModifyHeader() {
+        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
+        assertTrue(restHighLevelClient.modifyHeader(builder,
+            new BasicHeader("Content-Type", "application/json; Charset=UTF-16"), "Content-Type"));
+
+        assertThat(builder.getHeaders().stream().map(h -> h.getName() + "=>" + h.getValue()).collect(Collectors.joining(",")),
+            containsString("Content-Type=>application/vnd.elasticsearch+json; compatible-with=7; Charset=UTF-16"));
+
+        builder = RequestOptions.DEFAULT.toBuilder();
+        assertFalse(restHighLevelClient.modifyHeader(builder, new BasicHeader("Content-Type", "other"), "Content-Type"));
+
+        assertThat(builder.getHeaders().stream().map(h -> h.getName() + "=>" + h.getValue()).collect(Collectors.joining(",")),
+            equalTo(""));
+    }
+
+    public void testAddCompatibilityFor() {
+        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
+        Header entityHeader = new BasicHeader("Content-Type", "application/json");
+        String headerName = "Content-Type";
+
+        // No request headers, use entity header
+        assertTrue(restHighLevelClient.addCompatibilityFor(builder, entityHeader, headerName));
+        assertThat(builder.getHeaders().stream().map(h -> h.getName() + "=>" + h.getValue()).collect(Collectors.joining(",")),
+            containsString("Content-Type=>application/vnd.elasticsearch+json; compatible-with=7"));
+
+        // Request has a header, ignore entity header
+        builder = RequestOptions.DEFAULT.toBuilder().addHeader("Content-Type", "application/yaml Charset=UTF-32");
+        assertTrue(restHighLevelClient.addCompatibilityFor(builder, entityHeader, headerName));
+        assertThat(builder.getHeaders().stream().map(h -> h.getName() + "=>" + h.getValue()).collect(Collectors.joining(",")),
+            containsString("Content-Type=>application/vnd.elasticsearch+yaml; compatible-with=7 Charset=UTF-32"));
+
+        // Request has no headers, and no entity, no changes
+        builder = RequestOptions.DEFAULT.toBuilder();
+        assertFalse(restHighLevelClient.addCompatibilityFor(builder, null, headerName));
+        assertThat(builder.getHeaders().stream().map(h -> h.getName() + "=>" + h.getValue()).collect(Collectors.joining(",")),
+            equalTo(""));
+    }
+
+    public void testModifyForCompatibility() {
+        final Function<Request, String> allHeaders = r ->
+            r.getOptions().getHeaders().stream().map(h -> h.getName() + "=>" + h.getValue()).collect(Collectors.joining(","));
+
+        Request req = new Request("POST", "/");
+
+        restHighLevelClient.modifyRequestForCompatibility(req);
+
+        assertThat(allHeaders.apply(req), containsString(""));
+
+        // With an entity
+        req = new Request("POST", "/");
+        req.setEntity(new StringEntity("{}", ContentType.APPLICATION_JSON));
+        restHighLevelClient.modifyRequestForCompatibility(req);
+
+        assertThat(allHeaders.apply(req),
+            containsString("Content-Type=>application/vnd.elasticsearch+json; compatible-with=7; charset=UTF-8," +
+                "Accept=>application/vnd.elasticsearch+json; compatible-with=7"));
+
+        // With "Content-Type" headers already set
+        req = new Request("POST", "/");
+        req.setEntity(new StringEntity("{}", ContentType.TEXT_PLAIN));
+        req.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("Content-Type", "application/json; Charset=UTF-16"));
+        restHighLevelClient.modifyRequestForCompatibility(req);
+
+        assertThat(allHeaders.apply(req),
+            containsString("Content-Type=>application/vnd.elasticsearch+json; compatible-with=7; Charset=UTF-16," +
+                "Accept=>application/vnd.elasticsearch+json; compatible-with=7"));
+
+        // With "Content-Type" and "Accept" headers already set
+        req = new Request("POST", "/");
+        req.setEntity(new StringEntity("{}", ContentType.TEXT_PLAIN));
+        req.setOptions(RequestOptions.DEFAULT.toBuilder()
+            .addHeader("Content-Type", "application/json; Charset=UTF-16")
+            .addHeader("Accept", "application/yaml; Charset=UTF-32"));
+        // TODO: Fixme, ConcurrentModificationException here
+        restHighLevelClient.modifyRequestForCompatibility(req);
+
+        assertThat(allHeaders.apply(req),
+            containsString("Content-Type=>application/vnd.elasticsearch+json; compatible-with=7; Charset=UTF-16," +
+                "Accept=>application/vnd.elasticsearch+yaml; compatible-with=7; Charset=UTF-32"));
+
+    }
+
     private static void assertSyncMethod(Method method, String apiName, List<String> booleanReturnMethods) {
         //A few methods return a boolean rather than a response object
         if (apiName.equals("ping") || apiName.contains("exist") || booleanReturnMethods.contains(apiName)) {

client/rest/build.gradle

@@ -21,6 +21,7 @@ import org.elasticsearch.gradle.internal.conventions.precommit.LicenseHeadersTas
 
 apply plugin: 'elasticsearch.build'
 apply plugin: 'elasticsearch.publish'
+apply plugin: 'elasticsearch.internal-test-artifact'
 
 targetCompatibility = JavaVersion.VERSION_1_8
 sourceCompatibility = JavaVersion.VERSION_1_8