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(); + } + }