Merge branch '4.3.x'

Author: ryanjbaxter

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.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;
@@ -54,8 +58,14 @@ public class WebClientTransportClientFactory implements TransportClientFactory {
 
 	private final Supplier<WebClient.Builder> builderSupplier;
 
+	private final ConnectionProvider connectionProvider;
+
+	private final LoopResources loopResources;
+
 	public WebClientTransportClientFactory(Supplier<WebClient.Builder> builderSupplier) {
 		this.builderSupplier = builderSupplier;
+		this.connectionProvider = ConnectionProvider.create("eureka-webclient");
+		this.loopResources = LoopResources.create("eureka-webclient");
 	}
 
 	@Override
@@ -65,6 +75,11 @@ public EurekaHttpClient newClient(EurekaEndpoint endpoint) {
 		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());
 	}
 
@@ -113,6 +128,8 @@ private ExchangeFilterFunction http4XxErrorExchangeFilterFunction() {
 
 	@Override
 	public void shutdown() {
+		this.connectionProvider.dispose();
+		this.loopResources.dispose();
 	}
 
 }

spring-cloud-netflix-eureka-client/src/test/java/org/springframework/cloud/netflix/eureka/http/WebClientEurekaHttpClientTests.java

@@ -18,17 +18,25 @@
 
 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;
 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 @@ void setup() {
 		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();
+	}
+
 }