From 79b7b91a0352888086fedd8c4a2ca77ec47dbf77 Mon Sep 17 00:00:00 2001 From: Ryan Baxter Date: Wed, 25 Mar 2026 15:55:27 -0400 Subject: [PATCH] Use own connectionprovider and loopresources for WebClient MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When eureka.client.webclient.enabled=true is used with a reactive web server (Netty), the WebClient in WebClientEurekaHttpClient shares the same Reactor Netty event loop resources (LoopResources / ConnectionProvider) with the reactive web server, via Spring Boot's ReactorResourceFactory. During graceful shutdown, the sequence is: 1. EurekaServiceRegistry.deregister() sets the instance status to DOWN (but doesn't send the actual DELETE) 2. The reactive Netty web server shuts down, terminating its event loop 3. Later, during bean destruction, DiscoveryClient.shutdown() calls WebClientEurekaHttpClient.cancel() to send the DELETE deregistration request 4. cancel() calls Mono.block() on the WebClient, which tries to use the now-terminated Netty event loop → RejectedExecutionException: event executor terminated Fixes #4488 --- .../http/WebClientTransportClientFactory.java | 17 ++++++++++ .../http/WebClientEurekaHttpClientTests.java | 33 +++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/spring-cloud-netflix-eureka-client/src/main/java/org/springframework/cloud/netflix/eureka/http/WebClientTransportClientFactory.java b/spring-cloud-netflix-eureka-client/src/main/java/org/springframework/cloud/netflix/eureka/http/WebClientTransportClientFactory.java index 9761f9032..c50af80d1 100644 --- a/spring-cloud-netflix-eureka-client/src/main/java/org/springframework/cloud/netflix/eureka/http/WebClientTransportClientFactory.java +++ b/spring-cloud-netflix-eureka-client/src/main/java/org/springframework/cloud/netflix/eureka/http/WebClientTransportClientFactory.java @@ -24,9 +24,13 @@ import com.netflix.discovery.shared.transport.EurekaHttpClient; import com.netflix.discovery.shared.transport.TransportClientFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.netty.http.client.HttpClient; +import reactor.netty.resources.ConnectionProvider; +import reactor.netty.resources.LoopResources; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; +import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.http.codec.ClientCodecConfigurer; import org.springframework.http.codec.json.Jackson2JsonDecoder; import org.springframework.http.codec.json.Jackson2JsonEncoder; @@ -53,8 +57,14 @@ public class WebClientTransportClientFactory implements TransportClientFactory { private final Supplier builderSupplier; + private final ConnectionProvider connectionProvider; + + private final LoopResources loopResources; + public WebClientTransportClientFactory(Supplier builderSupplier) { this.builderSupplier = builderSupplier; + this.connectionProvider = ConnectionProvider.create("eureka-webclient"); + this.loopResources = LoopResources.create("eureka-webclient"); } @Override @@ -64,6 +74,11 @@ public class WebClientTransportClientFactory implements TransportClientFactory { setUrl(builder, endpoint.getServiceUrl()); setCodecs(builder); builder.filter(http4XxErrorExchangeFilterFunction()); + // Use dedicated Reactor Netty resources independent of the reactive web server + // to prevent RejectedExecutionException during graceful shutdown when the + // server's event loop terminates before DiscoveryClient deregisters. + builder.clientConnector( + new ReactorClientHttpConnector(HttpClient.create(this.connectionProvider).runOn(this.loopResources))); return new WebClientEurekaHttpClient(builder.build()); } @@ -110,6 +125,8 @@ public class WebClientTransportClientFactory implements TransportClientFactory { @Override public void shutdown() { + this.connectionProvider.dispose(); + this.loopResources.dispose(); } } diff --git a/spring-cloud-netflix-eureka-client/src/test/java/org/springframework/cloud/netflix/eureka/http/WebClientEurekaHttpClientTests.java b/spring-cloud-netflix-eureka-client/src/test/java/org/springframework/cloud/netflix/eureka/http/WebClientEurekaHttpClientTests.java index 49e409575..18c973955 100644 --- a/spring-cloud-netflix-eureka-client/src/test/java/org/springframework/cloud/netflix/eureka/http/WebClientEurekaHttpClientTests.java +++ b/spring-cloud-netflix-eureka-client/src/test/java/org/springframework/cloud/netflix/eureka/http/WebClientEurekaHttpClientTests.java @@ -18,7 +18,11 @@ package org.springframework.cloud.netflix.eureka.http; import com.netflix.appinfo.providers.EurekaConfigBasedInstanceInfoProvider; import com.netflix.discovery.shared.resolver.DefaultEndpoint; +import com.netflix.discovery.shared.transport.EurekaHttpClient; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.netty.http.client.HttpClient; +import reactor.netty.resources.LoopResources; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; @@ -26,9 +30,13 @@ import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; import org.springframework.cloud.commons.util.InetUtils; import org.springframework.cloud.netflix.eureka.EurekaInstanceConfigBean; +import org.springframework.http.HttpStatus; +import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.test.annotation.DirtiesContext; import org.springframework.web.reactive.function.client.WebClient; +import static org.assertj.core.api.Assertions.assertThat; + /** * @author Daniel Lavoie */ @@ -65,4 +73,29 @@ class WebClientEurekaHttpClientTests extends AbstractEurekaHttpClientTests { info = new EurekaConfigBasedInstanceInfoProvider(config).get(); } + @Test + void cancelSucceedsAfterSharedReactorResourcesDisposed() { + // Simulate the LoopResources shared between the reactive web server and a + // WebClient (as Spring Boot's ReactorResourceFactory provides by default) + LoopResources sharedResources = LoopResources.create("simulated-server"); + WebClient.Builder sharedBuilder = WebClient.builder() + .clientConnector(new ReactorClientHttpConnector(HttpClient.create().runOn(sharedResources))); + + WebClientTransportClientFactory factory = new WebClientTransportClientFactory(() -> sharedBuilder); + EurekaHttpClient client = factory.newClient(new DefaultEndpoint(serviceUrl)); + + // Simulate the reactive web server shutting down (terminates the shared event + // loop) + sharedResources.dispose(); + + // cancel() must succeed because the factory uses its own dedicated Reactor Netty + // resources that are independent of the shared (now disposed) web server + // resources. + // Without the fix this would throw RejectedExecutionException: event executor + // terminated + assertThat(client.cancel("test", "test").getStatusCode()).isEqualTo(HttpStatus.OK.value()); + + factory.shutdown(); + } + }