Reindex: Fix headers in reindex action (#26937)

The headers passed to reindex were skipped except for the last one. This
commit fixes the copying of the headers, as well as adds a base test
case for rest client builders to access the headers within the built
rest client.

relates #22976

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

@@ -50,6 +50,7 @@ import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -91,8 +92,9 @@ public class RestClient implements Closeable {
     private static final Log logger = LogFactory.getLog(RestClient.class);
 
     private final CloseableHttpAsyncClient client;
-    //we don't rely on default headers supported by HttpAsyncClient as those cannot be replaced
-    private final Header[] defaultHeaders;
+    // We don't rely on default headers supported by HttpAsyncClient as those cannot be replaced.
+    // These are package private for tests.
+    final List<Header> defaultHeaders;
     private final long maxRetryTimeoutMillis;
     private final String pathPrefix;
     private final AtomicInteger lastHostIndex = new AtomicInteger(0);
@@ -104,7 +106,7 @@ public class RestClient implements Closeable {
                HttpHost[] hosts, String pathPrefix, FailureListener failureListener) {
         this.client = client;
         this.maxRetryTimeoutMillis = maxRetryTimeoutMillis;
-        this.defaultHeaders = defaultHeaders;
+        this.defaultHeaders = Collections.unmodifiableList(Arrays.asList(defaultHeaders));
         this.failureListener = failureListener;
         this.pathPrefix = pathPrefix;
         setHosts(hosts);

modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportReindexAction.java

@@ -201,7 +201,7 @@ public class TransportReindexAction extends HandledTransportAction<ReindexReques
         Header[] clientHeaders = new Header[remoteInfo.getHeaders().size()];
         int i = 0;
         for (Map.Entry<String, String> header : remoteInfo.getHeaders().entrySet()) {
-            clientHeaders[i] = new BasicHeader(header.getKey(), header.getValue());
+            clientHeaders[i++] = new BasicHeader(header.getKey(), header.getValue());
         }
         return RestClient.builder(new HttpHost(remoteInfo.getHost(), remoteInfo.getPort(), remoteInfo.getScheme()))
                 .setDefaultHeaders(clientHeaders)

modules/reindex/src/test/java/org/elasticsearch/index/reindex/ReindexFromRemoteBuildRestClientTests.java

@@ -20,17 +20,21 @@
 package org.elasticsearch.index.reindex;
 
 import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilderTestCase;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.synchronizedList;
 import static org.hamcrest.Matchers.hasSize;
 
-public class ReindexFromRemoteBuildRestClientTests extends ESTestCase {
+public class ReindexFromRemoteBuildRestClientTests extends RestClientBuilderTestCase {
     public void testBuildRestClient() throws Exception {
         RemoteInfo remoteInfo = new RemoteInfo("https", "localhost", 9200, new BytesArray("ignored"), null, null, emptyMap(),
                 RemoteInfo.DEFAULT_SOCKET_TIMEOUT, RemoteInfo.DEFAULT_CONNECT_TIMEOUT);
@@ -48,4 +52,22 @@ public class ReindexFromRemoteBuildRestClientTests extends ESTestCase {
             client.close();
         }
     }
+
+    public void testHeaders() throws Exception {
+        Map<String, String> headers = new HashMap<>();
+        int numHeaders = randomIntBetween(1, 5);
+        for (int i = 0; i < numHeaders; ++i) {
+            headers.put("header" + i, Integer.toString(i));
+        }
+        RemoteInfo remoteInfo = new RemoteInfo("https", "localhost", 9200, new BytesArray("ignored"), null, null,
+            headers, RemoteInfo.DEFAULT_SOCKET_TIMEOUT, RemoteInfo.DEFAULT_CONNECT_TIMEOUT);
+        long taskId = randomLong();
+        List<Thread> threads = synchronizedList(new ArrayList<>());
+        RestClient client = TransportReindexAction.buildRestClient(remoteInfo, taskId, threads);
+        try {
+            assertHeaders(client, headers);
+        } finally {
+            client.close();
+        }
+    }
 }

test/framework/src/main/java/org/elasticsearch/client/RestClientBuilderTestCase.java

@@ -0,0 +1,48 @@
+/*
+ * 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 java.util.HashMap;
+import java.util.Map;
+
+import joptsimple.internal.Strings;
+import org.apache.http.Header;
+import org.elasticsearch.test.ESTestCase;
+
+/**
+ * A test case with access to internals of a RestClient.
+ */
+public abstract class RestClientBuilderTestCase extends ESTestCase {
+    /** Checks the given rest client has the provided default headers. */
+    public void assertHeaders(RestClient client, Map<String, String> expectedHeaders) {
+        expectedHeaders = new HashMap<>(expectedHeaders); // copy so we can remove as we check
+        for (Header header : client.defaultHeaders) {
+            String name = header.getName();
+            String expectedValue = expectedHeaders.remove(name);
+            if (expectedValue == null) {
+                fail("Found unexpected header in rest client: " + name);
+            }
+            assertEquals(expectedValue, header.getValue());
+        }
+        if (expectedHeaders.isEmpty() == false) {
+            fail("Missing expected headers in rest client: " + Strings.join(expectedHeaders.keySet(), ", "));
+        }
+    }
+}