reactor loadbalancer unable to locate service node after it's restarted

[paulluap]

Describe the bug

The reactor loadbalancer with health check (HealthCheckServiceInstanceListSupplier) cannot locate a service node after the node is restarted on a different port.

The following gateway properties are set:

#turn on the recommended ReactiveLoadBalancerClientFilter instead of ribbon
spring.cloud.loadbalancer.ribbon.enabled=false
# turns on withHealthChecks().withCaching
spring.cloud.loadbalancer.configurations=health-check

spring.application.name=demo-gateway
spring.cloud.gateway.discovery.locator.enabled=true

Steps to reproduce:

1. start consul, gateway in port 8080, a downstream service named demo-service, say on port 8082
2. client sends request to service via gateway:
   curl localhost:8080/demo-service/hello
   #response: hello
3. restart the service on port 8083, repeat 2, then the gateway always returns 503.

When debuging, it is found in the method:
org.springframework.cloud.loadbalancer.core.HealthCheckServiceInstanceListSupplier.healthCheckFlux(List<ServiceInstance>)
It always checks an obselete service instance (the one with port 8082), which is never alive after the restart.

spring boot and cloud version: 2.2.11.RELEASE, Hoxton.SR8

Sample

Minimun example to reproduce:
https://github.com/paulluap/reproduce-lb-failure

[paulluap]

I was able to workaround in my project by adding a custom refreshing instance list supplier that periodically fetches the instances from the discovery instance list supplier.

@Configuration
//TODO: apply this globally
@LoadBalancerClient(value="demo-service", configuration = CustomLoadBalancerConfiguration.class)
public class DemoServiceLBConfiguration { }

class CustomLoadBalancerConfiguration { 

	@Bean
	public ServiceInstanceListSupplier healthCheckDiscoveryClientServiceInstanceListSupplier(
			ConfigurableApplicationContext context, Environment env) {
		ReactiveDiscoveryClient discoveryClient = context
				.getBean(ReactiveDiscoveryClient.class);
		
		DiscoveryClientServiceInstanceListSupplier dsListSupplier = new DiscoveryClientServiceInstanceListSupplier(discoveryClient, env);
		return ServiceInstanceListSupplier
				.builder()
				//.withDiscoveryClient()
				.withBase(new RefresheableDSInstanceListSuppier(dsListSupplier))
				.withHealthChecks().withCaching().build(context);
	}
	
}

class RefresheableDSInstanceListSuppier extends DelegatingServiceInstanceListSupplier {

	public RefresheableDSInstanceListSuppier(ServiceInstanceListSupplier delegate) {
		super(delegate);
	}
	
	//this cannot handle RefreshRoutesEvent, so we rely on periodically reloading the service instances
	@Override
	public Flux<List<ServiceInstance>> get() {
		return Flux.defer(delegate).doOnNext(list->{
			System.out.println("refetched service instances from discovery client service instance list supplier");
		}).repeatWhen(reload->reload.delayElements(Duration.ofSeconds(30)));
	}

}

[OlgaMaciaszek]

@paulluap We suggest not to use LB caching together with the HealthCheckServiceInstanceListSupplier (see the docs), as it has it's own caching mechanism.
I realise that the default config should not turn the caching on because of that. Have created a separate issue for that and will fix it now.

Does the problem exist also if you remove withCaching() from your supplier config (by using the supplier builder and providing your own config in place of the default)?

[paulluap]

@OlgaMaciaszek I disabled cache by setting

spring.cloud.loadbalancer.cache.enabled=false

I think this has the same effect as removing withCaching in the builder call.

The issue is still reproduced.

If I understand correctly, I think the problem is in HealthCheckServiceInstanceListSupplier

Flux.defer(delegate)
				.delaySubscription(Duration.ofMillis(healthCheck.getInitialDelay()))
				.switchMap(serviceInstances -> healthCheckFlux(serviceInstances))

The source reactive stream produced by Flux.defer(delegate) does not continuously fire the latest service instances, though the lambda in switchMap ensures repeated health check on service instances.

As in my example using the consul blocking discovery client, the source service instance list is produced in essence by

class ConsulDiscoveryClient {
...
    Flux.just(delegate.getInstances(serviceId)

which would not emit new service instance once more, leaving down stream subscriptions seeing NOT up to date service instances.

[OlgaMaciaszek]

@paulluap I was able to reproduce it. Will provide a fix. Thanks.