From 25bdc281e5f8198d33be211f9dc7f566ad58242f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9E=97=E6=A1=89?= <161032298+014-code@users.noreply.github.com> Date: Tue, 14 Jul 2026 00:43:41 +0800 Subject: [PATCH] Fix async Rest5 client withHeaders() not seeing ThreadLocal auth Closes #3300 Signed-off-by: 014-code <2402143478@qq.com> (cherry picked from commit 7e70435f48db63f0ef70c45a8f7bedd6473725e3) --- .../client/elc/rest5_client/Rest5Clients.java | 22 ++++++++++--------- .../elasticsearch/client/RestClientsTest.java | 13 ++++++++++- 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/rest5_client/Rest5Clients.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/rest5_client/Rest5Clients.java index f0b17fb4a..fc6e06534 100644 --- a/src/main/java/org/springframework/data/elasticsearch/client/elc/rest5_client/Rest5Clients.java +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/rest5_client/Rest5Clients.java @@ -93,16 +93,18 @@ public final class Rest5Clients { throw new RuntimeException(e); } } - httpAsyncClientBuilder.addRequestInterceptorFirst((request, entity, context) -> { - clientConfiguration.getHeadersSupplier().get().forEach((header, values) -> { - // The accept and content-type headers are already put on the request, despite this being the first - // interceptor. - if ("Accept".equalsIgnoreCase(header) || " Content-Type".equalsIgnoreCase(header)) { - request.removeHeaders(header); - } - values.forEach(value -> request.addHeader(header, value)); - }); - }); + httpAsyncClientBuilder.addExecInterceptorFirst("es-rest5-client", + (request, entityProducer, scope, chain, asyncExecCallback) -> { + clientConfiguration.getHeadersSupplier().get().forEach((header, values) -> { + // The accept and content-type headers may already be put on the request, despite this being the + // first interceptor. + if ("Accept".equalsIgnoreCase(header) || "Content-Type".equalsIgnoreCase(header)) { + request.removeHeaders(header); + } + values.forEach(value -> request.addHeader(header, value)); + }); + chain.proceed(request, entityProducer, scope, asyncExecCallback); + }); // add httpclient configurator callbacks provided by the configuration for (ClientConfiguration.ClientConfigurationCallback clientConfigurer : clientConfiguration diff --git a/src/test/java/org/springframework/data/elasticsearch/client/RestClientsTest.java b/src/test/java/org/springframework/data/elasticsearch/client/RestClientsTest.java index eda7ed6a9..3ddfdac8c 100644 --- a/src/test/java/org/springframework/data/elasticsearch/client/RestClientsTest.java +++ b/src/test/java/org/springframework/data/elasticsearch/client/RestClientsTest.java @@ -32,6 +32,7 @@ import io.specto.hoverfly.junit5.api.HoverflyConfig; import java.io.IOException; import java.util.Arrays; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.stream.Stream; @@ -108,6 +109,8 @@ public class RestClientsTest { AtomicInteger connectionConfigurerCount = new AtomicInteger(0); AtomicInteger connectionManagerConfigurerCount = new AtomicInteger(0); AtomicInteger requestConfigurerCount = new AtomicInteger(0); + ThreadLocal threadLocal = ThreadLocal.withInitial(() -> null); + AtomicReference supplierThreadName = new AtomicReference<>(); ClientConfigurationBuilder configurationBuilder = new ClientConfigurationBuilder(); configurationBuilder // @@ -115,7 +118,9 @@ public class RestClientsTest { .withBasicAuth("user", "password") // .withDefaultHeaders(defaultHeaders) // .withHeaders(() -> { + supplierThreadName.set(Thread.currentThread().getName()); HttpHeaders httpHeaders = new HttpHeaders(); + httpHeaders.add("thread", threadLocal.get()); httpHeaders.add("supplied", "val0"); httpHeaders.add("supplied", "val" + supplierCount.getAndIncrement()); return httpHeaders; @@ -179,6 +184,7 @@ public class RestClientsTest { // do several calls to check that the headerSupplier provided values are set int startValue = clientUnderTest.usesInitialRequest() ? 2 : 1; for (int i = startValue; i <= startValue + 2; i++) { + threadLocal.set("local"); clientUnderTest.ping(); verify(headRequestedFor(urlEqualTo("/")) // @@ -189,10 +195,15 @@ public class RestClientsTest { .withHeader("supplied", new EqualToPattern("val0")) // // on the first call Elasticsearch does the version check and thus already increments the counter .withHeader("supplied", new EqualToPattern("val" + i)) // - .withHeader("supplied", including("val0", "val" + i))); + .withHeader("supplied", including("val0", "val" + i)) // + .withHeader("thread", new EqualToPattern("local"))); ; } + // the headers supplier must run on the calling thread, so a ThreadLocal set above is visible (#3300). + // this guards against future regressions that move the supplier back onto the async I/O thread. + assertThat(supplierThreadName.get()).isEqualTo(Thread.currentThread().getName()); + assertThat(restClientConfigurerCount).hasValue(clientUnderTestFactory.getExpectedRestClientConfigurerCalls()); assertThat(httpClientConfigurerCount).hasValue(1); assertThat(connectionConfigurerCount).hasValue(clientUnderTestFactory.getExpectedConnectionConfigurerCalls());