diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index bce6db46..71794eac 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -30,6 +30,9 @@ |spring.cloud.loadbalancer.health-check.initial-delay | `0` | Initial delay value for 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.health-check.refetch-instances | `false` | Indicates whether the instances should be refetched by the HealthCheckServiceInstanceListSupplier. This can be used if the instances can be updated and the underlying delegate does not provide an ongoing flux. +|spring.cloud.loadbalancer.health-check.refetch-instances-interval | `25s` | Interval for refetching available service instances. +|spring.cloud.loadbalancer.health-check.repeat-health-check | `true` | Indicates whether health checks should keep repeating. It might be useful to set it to false if periodically refetching the instances, as every refetch will also trigger a healthcheck. |spring.cloud.loadbalancer.hint | | Allows setting the value of hint that is passed on to the LoadBalancer request and can subsequently be used in {@link ReactiveLoadBalancer} implementations. |spring.cloud.loadbalancer.retry.avoid-previous-instance | `true` | Enables wrapping ServiceInstanceListSupplier beans with `RetryAwareServiceInstanceListSupplier` if Spring-Retry is in the classpath. |spring.cloud.loadbalancer.retry.backoff.enabled | `false` | Indicates whether Reactor Retry backoffs should be applied. diff --git a/docs/src/main/asciidoc/spring-cloud-commons.adoc b/docs/src/main/asciidoc/spring-cloud-commons.adoc index 8916b5a3..2fdf1554 100644 --- a/docs/src/main/asciidoc/spring-cloud-commons.adoc +++ b/docs/src/main/asciidoc/spring-cloud-commons.adoc @@ -929,15 +929,21 @@ TIP: This mechanism is particularly helpful while using the `SimpleDiscoveryClie clients backed by an actual Service Registry, it's not necessary to use, as we already get healthy instances after querying the external ServiceDiscovery. -TIP:: This supplier is also recommended for setups with a small number of instances per service +TIP: This supplier is also recommended for setups with a small number of instances per service in order to avoid retrying calls on a failing instance. +WARNING: If using any of the Service Discovery-backed suppliers, adding this health-check mechanism is usually not necessary, as we retrieve the health state of the instances directly +from the Service Registry. + +TIP: The `HealthCheckServiceInstanceListSupplier` relies on having updated instances provided by a delegate flux. In the rare cases when you want to use a delegate that does not refresh the instances, even though the list of instances may change (such as the `ReactiveDiscoveryClientServiceInstanceListSupplier` provided by us), you can set `spring.cloud.loadbalancer.health-check.refetch-instances` to `true` to have the instance list refreshed by the `HealthCheckServiceInstanceListSupplier`. You can then also adjust the refretch intervals by modifying the value of `spring.cloud.loadbalancer.health-check.refetch-instances-interval` and opt to disable the additional healthcheck repetitions by setting `spring.cloud.loadbalancer.repeat-health-check` to `fasle` as every instances refetch + will also trigger a healthcheck. + `HealthCheckServiceInstanceListSupplier` uses properties prefixed with `spring.cloud.loadbalancer.health-check`. You can set the `initialDelay` and `interval` for the scheduler. You can set the default path for the healthcheck URL by setting the value of the `spring.cloud.loadbalancer.health-check.path.default` property. You can also set a specific value for any given service by setting the value of the `spring.cloud.loadbalancer.health-check.path.[SERVICE_ID]` property, substituting `[SERVICE_ID]` with the correct ID of your service. If the path is not set, `/actuator/health` is used by default. -TIP:: If you rely on the default path (`/actuator/health`), make sure you add `spring-boot-starter-actuator` to your collaborator's dependencies, unless you are planning to add such an endpoint on your own. +TIP: If you rely on the default path (`/actuator/health`), make sure you add `spring-boot-starter-actuator` to your collaborator's dependencies, unless you are planning to add such an endpoint on your own. In order to use the health-check scheduler approach, you will have to instantiate a `HealthCheckServiceInstanceListSupplier` bean in a <>. @@ -962,7 +968,7 @@ public class CustomLoadBalancerConfiguration { } ---- -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`. +WARNING: `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`. [[spring-cloud-loadbalancer-hints]] === Spring Cloud LoadBalancer Hints 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 c25a4309..218bf534 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 @@ -89,8 +89,44 @@ public class LoadBalancerProperties { */ private Duration interval = Duration.ofSeconds(25); + /** + * Interval for refetching available service instances. + */ + private Duration refetchInstancesInterval = Duration.ofSeconds(25); + private Map path = new LinkedCaseInsensitiveMap<>(); + /** + * Indicates whether the instances should be refetched by the + * HealthCheckServiceInstanceListSupplier. This can be used if the + * instances can be updated and the underlying delegate does not provide an + * ongoing flux. + */ + private boolean refetchInstances = false; + + /** + * Indicates whether health checks should keep repeating. It might be useful to + * set it to false if periodically refetching the instances, as every + * refetch will also trigger a healthcheck. + */ + private boolean repeatHealthCheck = true; + + public boolean getRefetchInstances() { + return refetchInstances; + } + + public void setRefetchInstances(boolean refetchInstances) { + this.refetchInstances = refetchInstances; + } + + public boolean getRepeatHealthCheck() { + return repeatHealthCheck; + } + + public void setRepeatHealthCheck(boolean repeatHealthCheck) { + this.repeatHealthCheck = repeatHealthCheck; + } + public Duration getInitialDelay() { return initialDelay; } @@ -99,6 +135,14 @@ public class LoadBalancerProperties { this.initialDelay = initialDelay; } + public Duration getRefetchInstancesInterval() { + return refetchInstancesInterval; + } + + public void setRefetchInstancesInterval(Duration refetchInstancesInterval) { + this.refetchInstancesInterval = refetchInstancesInterval; + } + public Map getPath() { return path; } diff --git a/spring-cloud-context/src/main/java/org/springframework/cloud/autoconfigure/RefreshEndpointAutoConfiguration.java b/spring-cloud-context/src/main/java/org/springframework/cloud/autoconfigure/RefreshEndpointAutoConfiguration.java index b808a2f0..893879e0 100644 --- a/spring-cloud-context/src/main/java/org/springframework/cloud/autoconfigure/RefreshEndpointAutoConfiguration.java +++ b/spring-cloud-context/src/main/java/org/springframework/cloud/autoconfigure/RefreshEndpointAutoConfiguration.java @@ -16,6 +16,10 @@ package org.springframework.cloud.autoconfigure; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; + import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration; @@ -29,6 +33,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass; import org.springframework.cloud.context.properties.ConfigurationPropertiesRebinder; import org.springframework.cloud.context.refresh.ContextRefresher; +import org.springframework.cloud.context.restart.PauseHandler; import org.springframework.cloud.context.restart.RestartEndpoint; import org.springframework.cloud.context.scope.refresh.RefreshScope; import org.springframework.cloud.endpoint.RefreshEndpoint; @@ -36,6 +41,7 @@ import org.springframework.cloud.health.RefreshScopeHealthIndicator; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; +import org.springframework.integration.core.Pausable; import org.springframework.integration.monitor.IntegrationMBeanExporter; /** @@ -77,9 +83,17 @@ public class RefreshEndpointAutoConfiguration { @ConditionalOnClass(IntegrationMBeanExporter.class) class RestartEndpointWithIntegrationConfiguration { - @Autowired(required = false) private IntegrationMBeanExporter exporter; + RestartEndpointWithIntegrationConfiguration(@Autowired(required = false) IntegrationMBeanExporter exporter) { + this.exporter = exporter; + } + + @Bean + public PauseHandler integrationPauseHandler(ObjectProvider pausables) { + return new IntegrationPauseHandler(pausables.orderedStream().collect(Collectors.toList())); + } + @Bean @ConditionalOnAvailableEndpoint @ConditionalOnMissingBean @@ -91,6 +105,30 @@ class RestartEndpointWithIntegrationConfiguration { return endpoint; } + private class IntegrationPauseHandler implements PauseHandler { + + private List pausables = new ArrayList<>(); + + IntegrationPauseHandler(List pausables) { + this.pausables.addAll(pausables); + } + + @Override + public void pause() { + for (Pausable pausable : this.pausables) { + pausable.pause(); + } + } + + @Override + public void resume() { + for (int i = this.pausables.size(); i-- > 0;) { + this.pausables.get(i).resume(); + } + } + + } + } @Configuration(proxyBeanMethods = false) diff --git a/spring-cloud-context/src/main/java/org/springframework/cloud/context/restart/PauseHandler.java b/spring-cloud-context/src/main/java/org/springframework/cloud/context/restart/PauseHandler.java new file mode 100644 index 00000000..57f46473 --- /dev/null +++ b/spring-cloud-context/src/main/java/org/springframework/cloud/context/restart/PauseHandler.java @@ -0,0 +1,28 @@ +/* + * Copyright 2020-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.context.restart; + +/** + * @author Dave Syer + */ +public interface PauseHandler { + + void pause(); + + void resume(); + +} diff --git a/spring-cloud-context/src/main/java/org/springframework/cloud/context/restart/RestartEndpoint.java b/spring-cloud-context/src/main/java/org/springframework/cloud/context/restart/RestartEndpoint.java index 345e53ca..d0de5b51 100644 --- a/spring-cloud-context/src/main/java/org/springframework/cloud/context/restart/RestartEndpoint.java +++ b/spring-cloud-context/src/main/java/org/springframework/cloud/context/restart/RestartEndpoint.java @@ -19,10 +19,12 @@ package org.springframework.cloud.context.restart; import java.io.Closeable; import java.io.IOException; import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - +// import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.boot.SpringApplication; @@ -63,6 +65,8 @@ public class RestartEndpoint implements ApplicationListener pauseHandlers = Collections.emptyList(); + private long timeout; // @ManagedAttribute @@ -88,6 +92,8 @@ public class RestartEndpoint implements ApplicationListener 0;) { + PauseHandler handler = this.pauseHandlers.get(i); + handler.resume(); } } diff --git a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java index 5db173a1..303e681c 100644 --- a/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java +++ b/spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java @@ -25,6 +25,7 @@ import org.apache.commons.logging.LogFactory; import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.retry.Repeat; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; @@ -61,13 +62,18 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns public HealthCheckServiceInstanceListSupplier(ServiceInstanceListSupplier delegate, LoadBalancerProperties.HealthCheck healthCheck, WebClient webClient) { super(delegate); - this.healthCheck = healthCheck; defaultHealthCheckPath = healthCheck.getPath().getOrDefault("default", "/actuator/health"); this.webClient = webClient; - aliveInstancesReplay = Flux.defer(delegate).delaySubscription(healthCheck.getInitialDelay()) + this.healthCheck = healthCheck; + Repeat aliveInstancesReplayRepeat = Repeat + .onlyIf(repeatContext -> this.healthCheck.getRefetchInstances()) + .fixedBackoff(healthCheck.getRefetchInstancesInterval()); + Flux> aliveInstancesFlux = Flux.defer(delegate) .switchMap(serviceInstances -> healthCheckFlux(serviceInstances) .map(alive -> Collections.unmodifiableList(new ArrayList<>(alive)))) - .replay(1).refCount(1); + .repeatWhen(aliveInstancesReplayRepeat); + aliveInstancesReplay = aliveInstancesFlux.delaySubscription(healthCheck.getInitialDelay()).replay(1) + .refCount(1); } @Override @@ -80,6 +86,8 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns } protected Flux> healthCheckFlux(List instances) { + Repeat healthCheckFluxRepeat = Repeat.onlyIf(repeatContext -> healthCheck.getRepeatHealthCheck()) + .fixedBackoff(healthCheck.getInterval()); return Flux.defer(() -> { List> checks = new ArrayList<>(instances.size()); for (ServiceInstance instance : instances) { @@ -110,7 +118,7 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns result.add(alive); return result; }).defaultIfEmpty(result); - }).repeatWhen(restart -> restart.delayElements(healthCheck.getInterval())); + }).repeatWhen(healthCheckFluxRepeat); } @Override diff --git a/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplierTests.java b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplierTests.java index e23430a2..e222e476 100644 --- a/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplierTests.java +++ b/spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplierTests.java @@ -17,6 +17,7 @@ package org.springframework.cloud.loadbalancer.core; import java.time.Duration; +import java.util.Collections; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -47,6 +48,8 @@ import org.springframework.web.bind.annotation.RestController; import org.springframework.web.reactive.function.client.WebClient; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** * Tests for {@link HealthCheckServiceInstanceListSupplier}. @@ -136,11 +139,11 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2))); - HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class); + HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class); Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance1); Mockito.doReturn(Mono.just(false)).when(mock).isAlive(serviceInstance2); @@ -165,11 +168,11 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2))); - HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class); + HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class); Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance1); Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance2); @@ -195,11 +198,11 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2))); - HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class); + HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class); Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance1); Mockito.doReturn(Mono.error(new RuntimeException("boom"))).when(mock).isAlive(serviceInstance2); @@ -224,7 +227,7 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2))); listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) { @@ -251,7 +254,7 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1))); listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) { @@ -277,11 +280,11 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2))); - HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class); + HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class); Mockito.doReturn(Mono.just(false), Mono.just(true)).when(mock).isAlive(serviceInstance1); Mockito.doReturn(Mono.error(new RuntimeException("boom"))).when(mock).isAlive(serviceInstance2); @@ -306,11 +309,11 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1))); - HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class); + HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class); Mockito.when(mock.isAlive(serviceInstance1)).thenReturn(Mono.never(), Mono.just(true)); listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) { @@ -336,7 +339,7 @@ class HealthCheckServiceInstanceListSupplierTests { port, false); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Flux> instances = Flux.just(Lists.list(serviceInstance1)) .concatWith(Flux.just(Lists.list(serviceInstance1, serviceInstance2)) @@ -358,6 +361,33 @@ class HealthCheckServiceInstanceListSupplierTests { .thenCancel().verify(VERIFY_TIMEOUT); } + @Test + void shouldRefetchInstances() { + healthCheck.setInitialDelay(Duration.ofSeconds(1)); + healthCheck.setRepeatHealthCheck(false); + healthCheck.setRefetchInstancesInterval(Duration.ofSeconds(1)); + healthCheck.setRefetchInstances(true); + ServiceInstance serviceInstance1 = new DefaultServiceInstance("ignored-service-1", SERVICE_ID, "127.0.0.1", + port, false); + ServiceInstance serviceInstance2 = new DefaultServiceInstance("ignored-service-2", SERVICE_ID, "127.0.0.2", + port, false); + + StepVerifier.withVirtualTime(() -> { + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); + when(delegate.get()).thenReturn(Flux.just(Collections.singletonList(serviceInstance1))) + .thenReturn(Flux.just(Collections.singletonList(serviceInstance2))); + listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) { + @Override + protected Mono isAlive(ServiceInstance serviceInstance) { + return Mono.just(true); + } + }; + return listSupplier.get(); + }).expectSubscription().expectNoEvent(healthCheck.getInitialDelay()).expectNext(Lists.list(serviceInstance1)) + .thenAwait(healthCheck.getRefetchInstancesInterval()).expectNext(Lists.list(serviceInstance2)) + .thenCancel().verify(VERIFY_TIMEOUT); + } + @Test void shouldCacheResultIfAfterPropertiesSetInvoked() { healthCheck.setInitialDelay(Duration.ofSeconds(1)); @@ -367,7 +397,7 @@ class HealthCheckServiceInstanceListSupplierTests { AtomicInteger emitCounter = new AtomicInteger(); StepVerifier.withVirtualTime(() -> { - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID); Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1))); @@ -397,7 +427,7 @@ class HealthCheckServiceInstanceListSupplierTests { final AtomicInteger instancesCanceled = new AtomicInteger(); final AtomicBoolean subscribed = new AtomicBoolean(); - ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class); + ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class); Mockito.when(delegate.get()).thenReturn(Flux.>never() .doOnSubscribe(subscription -> subscribed.set(true)).doOnCancel(instancesCanceled::incrementAndGet));