Using case-insensitive headers (#1744)

The check for the X-elastic-product header is currently case sensitive. However header names are
supposed to be case-insensitive. This commit makes the check case sensitive.

Relates #1745

Author: masseyke

mr/src/main/java/org/elasticsearch/hadoop/rest/Response.java

@@ -46,5 +46,10 @@ public interface Response {
 
     boolean hasFailed();
 
-    Map<String, List<String>> headers();
+    /**
+     * Returns the headers with the given name
+     * @param headerName, case-insensitive
+     * @return The list of matching headers
+     */
+    List<String> getHeaders(String headerName);
 }

mr/src/main/java/org/elasticsearch/hadoop/rest/RestClient.java

@@ -780,7 +780,7 @@ public ClusterInfo mainInfo() {
                     LOG.warn("Could not verify server is Elasticsearch! Invalid main action response body format [build_flavor].");
                 }
 
-                List<String> productHeader = response.headers().get(ELASTIC_PRODUCT_HEADER);
+                List<String> productHeader = response.getHeaders(ELASTIC_PRODUCT_HEADER);
                 boolean validElasticsearchHeader = productHeader != null && productHeader.size() == 1 && productHeader.get(0).equals(ELASTIC_PRODUCT_HEADER_VALUE);
                 boolean verifyServer = (major.on(V_7_X) && major.parseMinorVersion(versionNumber) >= 14) || major.onOrAfter(V_8_X);
                 if (validElasticsearchHeader == false) {

mr/src/main/java/org/elasticsearch/hadoop/rest/SimpleResponse.java

@@ -22,6 +22,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 public class SimpleResponse implements Response {
 
@@ -102,7 +103,8 @@ public boolean hasFailed() {
     }
 
     @Override
-    public Map<String, List<String>> headers() {
-        return headers;
+    public List<String> getHeaders(String headerName) {
+        return headers.entrySet().stream().filter(entry -> entry.getKey().equalsIgnoreCase(headerName)).map(Map.Entry::getValue)
+                .flatMap(List::stream).collect(Collectors.toList());
     }
 }

mr/src/test/java/org/elasticsearch/hadoop/rest/RestClientTest.java

@@ -35,6 +35,7 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
@@ -440,7 +441,7 @@ public void testMainInfo() {
 
         NetworkClient mock = Mockito.mock(NetworkClient.class);
         Map<String, List<String>> headers = new HashMap<>();
-        headers.computeIfAbsent(RestClient.ELASTIC_PRODUCT_HEADER, k -> new ArrayList<>()).add(RestClient.ELASTIC_PRODUCT_HEADER_VALUE);
+        headers.computeIfAbsent(RestClient.ELASTIC_PRODUCT_HEADER.toLowerCase(Locale.ROOT), k -> new ArrayList<>()).add(RestClient.ELASTIC_PRODUCT_HEADER_VALUE);
         Mockito.when(mock.execute(Mockito.any(SimpleRequest.class), Mockito.eq(true)))
                 .thenReturn(new SimpleResponse(201, new FastByteArrayInputStream(new BytesArray(response)), "localhost:9200", headers));