Set thread name used by REST client (#103160)

By default the REST client uses a thread factory which names its threads
with the generic pattern `I/O dispatcher %d`. This commit adds the
prefix `elasticsearch-rest-client-`, and a client-instance-specific ID,
to the name of these threads to make them easier to identify.

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

@@ -37,6 +37,8 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Objects;
 import java.util.Properties;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
 
 import javax.net.ssl.SSLContext;
 
@@ -51,6 +53,9 @@ public final class RestClientBuilder {
     public static final int DEFAULT_MAX_CONN_PER_ROUTE = 10;
     public static final int DEFAULT_MAX_CONN_TOTAL = 30;
 
+    static final String THREAD_NAME_PREFIX = "elasticsearch-rest-client-";
+    private static final String THREAD_NAME_FORMAT = THREAD_NAME_PREFIX + "%d-thread-%d";
+
     public static final String VERSION;
     static final String META_HEADER_NAME = "X-Elastic-Client-Meta";
     static final String META_HEADER_VALUE;
@@ -298,6 +303,24 @@ public final class RestClientBuilder {
         return restClient;
     }
 
+    /**
+     * Similar to {@code org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor.DefaultThreadFactory} but with better thread names.
+     */
+    private static class RestClientThreadFactory implements ThreadFactory {
+        private static final AtomicLong CLIENT_THREAD_POOL_ID_GENERATOR = new AtomicLong();
+
+        private final long clientThreadPoolId = CLIENT_THREAD_POOL_ID_GENERATOR.getAndIncrement(); // 0-based
+        private final AtomicLong clientThreadId = new AtomicLong();
+
+        @Override
+        public Thread newThread(Runnable runnable) {
+            return new Thread(
+                runnable,
+                String.format(Locale.ROOT, THREAD_NAME_FORMAT, clientThreadPoolId, clientThreadId.incrementAndGet()) // 1-based
+            );
+        }
+    }
+
     private CloseableHttpAsyncClient createHttpClient() {
         // default timeouts are all infinite
         RequestConfig.Builder requestConfigBuilder = RequestConfig.custom()
@@ -315,7 +338,8 @@ public final class RestClientBuilder {
                 .setMaxConnTotal(DEFAULT_MAX_CONN_TOTAL)
                 .setSSLContext(SSLContext.getDefault())
                 .setUserAgent(USER_AGENT_HEADER_VALUE)
-                .setTargetAuthenticationStrategy(new PersistentCredentialsAuthenticationStrategy());
+                .setTargetAuthenticationStrategy(new PersistentCredentialsAuthenticationStrategy())
+                .setThreadFactory(new RestClientThreadFactory());
             if (httpClientConfigCallback != null) {
                 httpClientBuilder = httpClientConfigCallback.customizeHttpClient(httpClientBuilder);
             }

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

@@ -42,15 +42,21 @@ import java.security.PrivilegedAction;
 import java.security.cert.Certificate;
 import java.security.cert.CertificateFactory;
 import java.security.spec.PKCS8EncodedKeySpec;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLHandshakeException;
 import javax.net.ssl.TrustManagerFactory;
 
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 /**
@@ -105,6 +111,40 @@ public class RestClientBuilderIntegTests extends RestClientTestCase {
         }
     }
 
+    public void testBuilderSetsThreadName() throws Exception {
+        assumeFalse("https://github.com/elastic/elasticsearch/issues/49094", inFipsJvm());
+        final SSLContext defaultSSLContext = SSLContext.getDefault();
+        try {
+            SSLContext.setDefault(getSslContext());
+            try (RestClient client = buildRestClient()) {
+                final CountDownLatch latch = new CountDownLatch(1);
+                client.performRequestAsync(new Request("GET", "/"), new ResponseListener() {
+                    @Override
+                    public void onSuccess(Response response) {
+                        assertThat(
+                            Thread.currentThread().getName(),
+                            allOf(
+                                startsWith(RestClientBuilder.THREAD_NAME_PREFIX),
+                                containsString("elasticsearch"),
+                                containsString("rest-client")
+                            )
+                        );
+                        assertEquals(200, response.getStatusLine().getStatusCode());
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void onFailure(Exception exception) {
+                        throw new AssertionError("unexpected", exception);
+                    }
+                });
+                assertTrue(latch.await(10, TimeUnit.SECONDS));
+            }
+        } finally {
+            SSLContext.setDefault(defaultSSLContext);
+        }
+    }
+
     private RestClient buildRestClient() {
         InetSocketAddress address = httpsServer.getAddress();
         return RestClient.builder(new HttpHost(address.getHostString(), address.getPort(), "https")).build();

docs/changelog/103160.yaml

@@ -0,0 +1,5 @@
+pr: 103160
+summary: Set thread name used by REST client
+area: Java Low Level REST Client
+type: enhancement
+issues: []