From 587f5f568a88c70cc5b938eaadb5c85711a82473 Mon Sep 17 00:00:00 2001 From: Olga Maciaszek-Sharma Date: Tue, 26 May 2020 09:35:09 -0500 Subject: [PATCH 1/2] Gh 760 health check with cache new (#765) * Cache first element of service instance list flux. * Invoke destroy() and afterPropertiesSet() in non-bean ServiceInstanceListSupplier delegates. * Fix return updated instances. * Fix return updated instances. (cherry picked from commit 88b2f0e8695ec4a3b46480ca02646ce1d9d265ac) --- .../CachingServiceInstanceListSupplier.java | 2 +- ...DelegatingServiceInstanceListSupplier.java | 21 ++- ...veryClientServiceInstanceListSupplier.java | 11 +- ...chingServiceInstanceListSupplierTests.java | 162 ++++++++++++++++++ ...lientServiceInstanceListSupplierTests.java | 132 ++++++++++++++ src/checkstyle/checkstyle-suppressions.xml | 1 + 6 files changed, 321 insertions(+), 8 deletions(-) create mode 100644 spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplierTests.java create mode 100644 spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java diff --git a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java index aecca695..a9fcc362 100644 --- a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java +++ b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplier.java @@ -69,7 +69,7 @@ public class CachingServiceInstanceListSupplier return Mono.empty(); } return Flux.just(list).materialize().collectList(); - }, delegate.getServiceId()).onCacheMissResume(delegate) + }, delegate.getServiceId()).onCacheMissResume(delegate.get().take(1)) .andWriteWith((key, signals) -> Flux.fromIterable(signals).dematerialize() .doOnNext(instances -> { Cache cache = cacheManager diff --git a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DelegatingServiceInstanceListSupplier.java b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DelegatingServiceInstanceListSupplier.java index 7cc9d53f..50ba6a18 100644 --- a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DelegatingServiceInstanceListSupplier.java +++ b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DelegatingServiceInstanceListSupplier.java @@ -16,6 +16,8 @@ package org.springframework.cloud.loadbalancer.core; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; /** @@ -23,11 +25,12 @@ import org.springframework.util.Assert; * {@link ServiceInstanceListSupplier} instance underneath. * * @author Spencer Gibb + * @author Olga Maciaszek-Sharma */ public abstract class DelegatingServiceInstanceListSupplier - implements ServiceInstanceListSupplier { + implements ServiceInstanceListSupplier, InitializingBean, DisposableBean { - private final ServiceInstanceListSupplier delegate; + protected final ServiceInstanceListSupplier delegate; public DelegatingServiceInstanceListSupplier(ServiceInstanceListSupplier delegate) { Assert.notNull(delegate, "delegate may not be null"); @@ -43,4 +46,18 @@ public abstract class DelegatingServiceInstanceListSupplier return this.delegate.getServiceId(); } + @Override + public void afterPropertiesSet() throws Exception { + if (delegate instanceof InitializingBean) { + ((InitializingBean) delegate).afterPropertiesSet(); + } + } + + @Override + public void destroy() throws Exception { + if (delegate instanceof DisposableBean) { + ((DisposableBean) delegate).destroy(); + } + } + } diff --git a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java index 9e5b5898..362387e0 100644 --- a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java +++ b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java @@ -41,20 +41,21 @@ public class DiscoveryClientServiceInstanceListSupplier private final String serviceId; - private final Flux serviceInstances; + private final Flux> serviceInstances; public DiscoveryClientServiceInstanceListSupplier(DiscoveryClient delegate, Environment environment) { this.serviceId = environment.getProperty(PROPERTY_NAME); this.serviceInstances = Flux - .defer(() -> Flux.fromIterable(delegate.getInstances(serviceId))) - .subscribeOn(Schedulers.boundedElastic()); + .defer(() -> Flux.fromIterable(delegate.getInstances(serviceId)) + .collectList().flux().subscribeOn(Schedulers.boundedElastic())); } public DiscoveryClientServiceInstanceListSupplier(ReactiveDiscoveryClient delegate, Environment environment) { this.serviceId = environment.getProperty(PROPERTY_NAME); - this.serviceInstances = delegate.getInstances(serviceId); + this.serviceInstances = Flux + .defer(() -> delegate.getInstances(serviceId).collectList().flux()); } @Override @@ -64,7 +65,7 @@ public class DiscoveryClientServiceInstanceListSupplier @Override public Flux> get() { - return serviceInstances.collectList().flux(); + return serviceInstances; } } diff --git a/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplierTests.java b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplierTests.java new file mode 100644 index 00000000..6b3e51a3 --- /dev/null +++ b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/CachingServiceInstanceListSupplierTests.java @@ -0,0 +1,162 @@ +/* + * Copyright 2012-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.loadbalancer.core; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.client.DefaultServiceInstance; +import org.springframework.cloud.client.ServiceInstance; +import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient; +import org.springframework.cloud.client.loadbalancer.reactive.LoadBalancerProperties; +import org.springframework.cloud.loadbalancer.blocking.client.BlockingLoadBalancerClient; +import org.springframework.cloud.loadbalancer.cache.LoadBalancerCacheManager; +import org.springframework.cloud.loadbalancer.config.LoadBalancerCacheAutoConfiguration; +import org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.junit.jupiter.SpringExtension; +import org.springframework.web.reactive.function.client.WebClient; + +import static java.time.Duration.ofMillis; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; + +/** + * Tests for {@link CachingServiceInstanceListSupplier}. + * + * @author Olga Maciaszek-Sharma + */ +@SpringBootTest(classes = CachingServiceInstanceListSupplierTests.TestConfig.class) +@ExtendWith(SpringExtension.class) +class CachingServiceInstanceListSupplierTests { + + public static final String SERVICE_ID = "test"; + + static { + System.setProperty("loadbalancer.client.name", SERVICE_ID); + } + + @Autowired + BlockingLoadBalancerClient blockingLoadBalancerClient; + + private static DefaultServiceInstance instance(String host, boolean secure) { + return new DefaultServiceInstance(SERVICE_ID, SERVICE_ID, host, 80, secure); + } + + @Test + void shouldNotHangOnCachingWhenDelegateReturnsInfiniteStream() { + assertTimeoutPreemptively(ofMillis(500), () -> { + blockingLoadBalancerClient.choose(SERVICE_ID); + }); + + } + + @Configuration(proxyBeanMethods = false) + @Import(LoadBalancerCacheAutoConfiguration.class) + protected static class TestConfig { + + @Bean + public ReactiveDiscoveryClient reactiveDiscoveryClient() { + return new ReactiveDiscoveryClient() { + @Override + public String description() { + return SERVICE_ID; + } + + @Override + public Flux getInstances(String serviceId) { + return Flux.just(instance("1host", false), + instance("2host-secure", true)); + } + + @Override + public Flux getServices() { + return Flux.just(SERVICE_ID); + } + }; + } + + @Bean + ReactorLoadBalancer reactorLoadBalancer( + ObjectProvider provider) { + return new RoundRobinLoadBalancer(provider, SERVICE_ID); + } + + @Bean + LoadBalancerClientFactory loadBalancerClientFactory() { + return new LoadBalancerClientFactory(); + } + + @Bean + BlockingLoadBalancerClient blockingLoadBalancerClient( + LoadBalancerClientFactory loadBalancerClientFactory) { + return new BlockingLoadBalancerClient(loadBalancerClientFactory); + } + + @Bean + public LoadBalancerProperties loadBalancerProperties() { + return new LoadBalancerProperties(); + } + + @Bean + public WebClient.Builder webClientBuilder() { + return WebClient.builder(); + } + + @Bean + ServiceInstanceListSupplier supplier(ConfigurableApplicationContext context, + ReactiveDiscoveryClient discoveryClient, + LoadBalancerProperties loadBalancerProperties, + WebClient.Builder webClientBuilder) { + DiscoveryClientServiceInstanceListSupplier firstDelegate = new DiscoveryClientServiceInstanceListSupplier( + discoveryClient, context.getEnvironment()); + HealthCheckServiceInstanceListSupplier delegate = new TestHealthCheckServiceInstanceListSupplier( + firstDelegate, loadBalancerProperties.getHealthCheck(), + webClientBuilder.build()); + delegate.afterPropertiesSet(); + ObjectProvider cacheManagerProvider = context + .getBeanProvider(LoadBalancerCacheManager.class); + return new CachingServiceInstanceListSupplier(delegate, + cacheManagerProvider.getIfAvailable()); + } + + private static class TestHealthCheckServiceInstanceListSupplier + extends HealthCheckServiceInstanceListSupplier { + + TestHealthCheckServiceInstanceListSupplier( + ServiceInstanceListSupplier delegate, + LoadBalancerProperties.HealthCheck healthCheck, WebClient webClient) { + super(delegate, healthCheck, webClient); + } + + @Override + protected Mono isAlive(ServiceInstance serviceInstance) { + return Mono.just(true); + } + + } + + } + +} diff --git a/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java new file mode 100644 index 00000000..e7f4ddea --- /dev/null +++ b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java @@ -0,0 +1,132 @@ +/* + * Copyright 2012-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.loadbalancer.core; + +import org.assertj.core.util.Lists; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.springframework.cloud.client.DefaultServiceInstance; +import org.springframework.cloud.client.discovery.DiscoveryClient; +import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient; +import org.springframework.mock.env.MockEnvironment; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Tests for {@link DiscoveryClientServiceInstanceListSupplier}. + * + * @author Olga Maciaszek-Sharma + */ +class DiscoveryClientServiceInstanceListSupplierTests { + + private static final String SERVICE_ID = "test"; + + private final MockEnvironment environment = new MockEnvironment(); + + private final ReactiveDiscoveryClient reactiveDiscoveryClient = mock( + ReactiveDiscoveryClient.class); + + private final DiscoveryClient discoveryClient = mock(DiscoveryClient.class); + + private DiscoveryClientServiceInstanceListSupplier supplier; + + private static DefaultServiceInstance instance(String host, boolean secure) { + return new DefaultServiceInstance(SERVICE_ID, SERVICE_ID, host, 80, secure); + } + + @BeforeEach + void setUp() { + environment.setProperty("loadbalancer.client.name", SERVICE_ID); + } + + @Test + void shouldReturnRetrievedInstances() { + when(reactiveDiscoveryClient.getInstances(SERVICE_ID)).thenReturn( + Flux.just(instance("1host", false), instance("2host-secure", true))); + + StepVerifier.withVirtualTime(() -> { + supplier = new DiscoveryClientServiceInstanceListSupplier( + reactiveDiscoveryClient, environment); + return supplier.get(); + }).expectSubscription().expectNext( + Lists.list(instance("1host", false), instance("2host-secure", true))) + .thenCancel().verify(); + } + + @Test + void shouldUpdateReturnRetrievedInstances() { + when(reactiveDiscoveryClient.getInstances(SERVICE_ID)).thenReturn( + Flux.just(instance("1host", false), instance("2host-secure", true))); + supplier = new DiscoveryClientServiceInstanceListSupplier(reactiveDiscoveryClient, + environment); + + StepVerifier.withVirtualTime(() -> supplier.get()).expectSubscription() + .expectNext(Lists.list(instance("1host", false), + instance("2host-secure", true))) + .thenCancel().verify(); + + when(reactiveDiscoveryClient.getInstances(SERVICE_ID)) + .thenReturn(Flux.just(instance("1host", false), + instance("2host-secure", true), instance("3host", false))); + + StepVerifier.withVirtualTime(() -> supplier.get()).expectSubscription() + .expectNext(Lists.list(instance("1host", false), + instance("2host-secure", true), instance("3host", false))) + .thenCancel().verify(); + } + + @Test + void shouldReturnRetrievedInstancesBlockingClient() { + StepVerifier.withVirtualTime(() -> { + when(discoveryClient.getInstances(SERVICE_ID)).thenReturn( + Lists.list(instance("1host", false), instance("2host-secure", true))); + + supplier = new DiscoveryClientServiceInstanceListSupplier(discoveryClient, + environment); + return supplier.get(); + }).expectSubscription().expectNext( + Lists.list(instance("1host", false), instance("2host-secure", true))) + .thenCancel().verify(); + } + + @Test + void shouldUpdateReturnRetrievedInstancesBlockingClient() { + when(discoveryClient.getInstances(SERVICE_ID)).thenReturn( + Lists.list(instance("1host", false), instance("2host-secure", true))); + supplier = new DiscoveryClientServiceInstanceListSupplier(discoveryClient, + environment); + + StepVerifier.withVirtualTime(() -> supplier.get()).expectSubscription() + .expectNext(Lists.list(instance("1host", false), + instance("2host-secure", true))) + .thenCancel().verify(); + + when(discoveryClient.getInstances(SERVICE_ID)) + .thenReturn(Lists.list(instance("1host", false), + instance("2host-secure", true), instance("3host", false))); + + StepVerifier.withVirtualTime(() -> supplier.get()).expectSubscription() + .expectNext(Lists.list(instance("1host", false), + instance("2host-secure", true), instance("3host", false))) + .thenCancel().verify(); + } + +} diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index 2a924e7e..20341623 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -14,4 +14,5 @@ + From 87e5d7a62bd99d3bc70906ccd60429210b841cbd Mon Sep 17 00:00:00 2001 From: Olga Maciaszek-Sharma Date: Wed, 27 May 2020 09:13:55 -0500 Subject: [PATCH 2/2] Fix execution thread for blocking and adjusts timing * Fix execution thread for blocking DiscoveryClientServiceInstanceListSupplier. * Desynchronise HealthCheck and Cache. Add info about using HealthCheck without Cache to docs. See gh-760 --- docs/src/main/asciidoc/_configprops.adoc | 4 +-- .../main/asciidoc/spring-cloud-commons.adoc | 7 +++--- .../reactive/LoadBalancerProperties.java | 2 +- .../config/SimpleBootstrapPropertySource.java | 1 + .../cache/LoadBalancerCacheProperties.java | 2 +- ...veryClientServiceInstanceListSupplier.java | 4 +-- ...lientServiceInstanceListSupplierTests.java | 25 ++++++++----------- 7 files changed, 22 insertions(+), 23 deletions(-) diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index 675556d9..e4bb6464 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -29,9 +29,9 @@ |spring.cloud.inetutils.use-only-site-local-interfaces | false | Whether to use only interfaces with site local addresses. See {@link InetAddress#isSiteLocalAddress()} for more details. |spring.cloud.loadbalancer.cache.caffeine.spec | | The spec to use to create caches. See CaffeineSpec for more details on the spec format. |spring.cloud.loadbalancer.cache.capacity | 256 | Initial cache capacity expressed as int. -|spring.cloud.loadbalancer.cache.ttl | 30s | Time To Live - time counted from writing of the record, after which cache entries are expired, expressed as a {@link Duration}. The property {@link String} has to be in keeping with the appropriate syntax as specified in Spring Boot StringToDurationConverter. @see StringToDurationConverter.java +|spring.cloud.loadbalancer.cache.ttl | 35s | Time To Live - time counted from writing of the record, after which cache entries are expired, expressed as a {@link Duration}. The property {@link String} has to be in keeping with the appropriate syntax as specified in Spring Boot StringToDurationConverter. @see StringToDurationConverter.java |spring.cloud.loadbalancer.health-check.initial-delay | 0 | Initial delay value for the HealthCheck scheduler. -|spring.cloud.loadbalancer.health-check.interval | 30s | Interval for rerunning the HealthCheck scheduler. +|spring.cloud.loadbalancer.health-check.interval | 25s | Interval for rerunning the HealthCheck scheduler. |spring.cloud.loadbalancer.health-check.path | | |spring.cloud.loadbalancer.retry.enabled | true | |spring.cloud.loadbalancer.ribbon.enabled | true | Causes `RibbonLoadBalancerClient` to be used by default. diff --git a/docs/src/main/asciidoc/spring-cloud-commons.adoc b/docs/src/main/asciidoc/spring-cloud-commons.adoc index 0f6b3ad3..54ec1b9f 100644 --- a/docs/src/main/asciidoc/spring-cloud-commons.adoc +++ b/docs/src/main/asciidoc/spring-cloud-commons.adoc @@ -882,7 +882,7 @@ You can set your own `ttl` value (the time after write after which entries shoul as the value of the `spring.cloud.loadbalancer.cache.ttl` property. You can also set your own LoadBalancer cache initial capacity by setting the value of the `spring.cloud.loadbalancer.cache.capacity` property. -The default setup includes `ttl` set to 30 seconds and the default `initialCapacity` is `256`. +The default setup includes `ttl` set to 35 seconds and the default `initialCapacity` is `256`. You can also altogether disable loadBalancer caching by setting the value of `spring.cloud.loadbalancer.cache.enabled` to `false`. @@ -951,7 +951,7 @@ We suggest passing a `DiscoveryClientServiceInstanceListSupplier` delegate in th You could use this sample configuration to set it up: -[[zoned-based-custom-loadbalancer-configuration]] +[[health-check-based-custom-loadbalancer-configuration]] [source,java,indent=0] ---- public class CustomLoadBalancerConfiguration { @@ -962,12 +962,13 @@ public class CustomLoadBalancerConfiguration { return ServiceInstanceListSupplier.builder() .withDiscoveryClient() .withHealthChecks() - .withCaching() .build(context); } } ---- +NOTE:: `HealthCheckServiceInstanceListSupplier` has its own caching mechanism based on Reactor Flux `replay()`, therefore, if it's being used, you may want to skip wrapping that supplier with `CachingServiceInstanceListSupplier`. + TIP:: In order to make working on your own LoadBalancer configuration easier, we have added a `builder()` method to the `ServiceInstanceListSupplier` class. TIP:: You can also use our alternative predefined configurations in place of the default ones by setting the value of `spring.cloud.loadbalancer.configurations` property to `zone-preference` to use `ZonePreferenceServiceInstanceListSupplier` with caching or to `health-check` to use `HealthCheckServiceInstanceListSupplier` with caching. diff --git a/spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerProperties.java b/spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerProperties.java index ca8da2f0..6c6ea92b 100644 --- a/spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerProperties.java +++ b/spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerProperties.java @@ -54,7 +54,7 @@ public class LoadBalancerProperties { /** * Interval for rerunning the HealthCheck scheduler. */ - private Duration interval = Duration.ofSeconds(30); + private Duration interval = Duration.ofSeconds(25); private Map path = new LinkedCaseInsensitiveMap<>(); diff --git a/spring-cloud-context/src/main/java/org/springframework/cloud/bootstrap/config/SimpleBootstrapPropertySource.java b/spring-cloud-context/src/main/java/org/springframework/cloud/bootstrap/config/SimpleBootstrapPropertySource.java index 1f702999..3d17653b 100644 --- a/spring-cloud-context/src/main/java/org/springframework/cloud/bootstrap/config/SimpleBootstrapPropertySource.java +++ b/spring-cloud-context/src/main/java/org/springframework/cloud/bootstrap/config/SimpleBootstrapPropertySource.java @@ -22,6 +22,7 @@ import static org.springframework.cloud.bootstrap.config.PropertySourceBootstrap /** * Simple, non-enumerable PropertySource wrapper. + * * @author Ryan Baxter */ public class SimpleBootstrapPropertySource extends PropertySource { diff --git a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/cache/LoadBalancerCacheProperties.java b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/cache/LoadBalancerCacheProperties.java index b67647aa..426c9382 100644 --- a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/cache/LoadBalancerCacheProperties.java +++ b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/cache/LoadBalancerCacheProperties.java @@ -39,7 +39,7 @@ public class LoadBalancerCacheProperties { * @see StringToDurationConverter.java */ - private Duration ttl = Duration.ofSeconds(30); + private Duration ttl = Duration.ofSeconds(35); /** * Initial cache capacity expressed as int. diff --git a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java index 362387e0..cbc4453f 100644 --- a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java +++ b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplier.java @@ -47,8 +47,8 @@ public class DiscoveryClientServiceInstanceListSupplier Environment environment) { this.serviceId = environment.getProperty(PROPERTY_NAME); this.serviceInstances = Flux - .defer(() -> Flux.fromIterable(delegate.getInstances(serviceId)) - .collectList().flux().subscribeOn(Schedulers.boundedElastic())); + .defer(() -> Flux.just(delegate.getInstances(serviceId))) + .subscribeOn(Schedulers.boundedElastic()); } public DiscoveryClientServiceInstanceListSupplier(ReactiveDiscoveryClient delegate, diff --git a/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java index e7f4ddea..4d102607 100644 --- a/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java +++ b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/DiscoveryClientServiceInstanceListSupplierTests.java @@ -109,21 +109,18 @@ class DiscoveryClientServiceInstanceListSupplierTests { @Test void shouldUpdateReturnRetrievedInstancesBlockingClient() { - when(discoveryClient.getInstances(SERVICE_ID)).thenReturn( - Lists.list(instance("1host", false), instance("2host-secure", true))); - supplier = new DiscoveryClientServiceInstanceListSupplier(discoveryClient, - environment); + StepVerifier.withVirtualTime(() -> { + when(discoveryClient.getInstances(SERVICE_ID)).thenReturn( + Lists.list(instance("1host", false), instance("2host-secure", true))); + supplier = new DiscoveryClientServiceInstanceListSupplier(discoveryClient, + environment); + supplier.get(); - StepVerifier.withVirtualTime(() -> supplier.get()).expectSubscription() - .expectNext(Lists.list(instance("1host", false), - instance("2host-secure", true))) - .thenCancel().verify(); - - when(discoveryClient.getInstances(SERVICE_ID)) - .thenReturn(Lists.list(instance("1host", false), - instance("2host-secure", true), instance("3host", false))); - - StepVerifier.withVirtualTime(() -> supplier.get()).expectSubscription() + when(discoveryClient.getInstances(SERVICE_ID)) + .thenReturn(Lists.list(instance("1host", false), + instance("2host-secure", true), instance("3host", false))); + return supplier.get(); + }).expectSubscription() .expectNext(Lists.list(instance("1host", false), instance("2host-secure", true), instance("3host", false))) .thenCancel().verify();