Fix reactive connection handling.

Original Pull Request #1766
Closes #1759

Author: sothawo

src/main/java/org/springframework/data/elasticsearch/client/reactive/DefaultReactiveElasticsearchClient.java

@@ -541,8 +541,7 @@ public <T> Mono<T> execute(ReactiveElasticsearchClientCallback<T> callback) {
 				.flatMap(callback::doWithClient) //
 				.onErrorResume(throwable -> {
 
-					if (throwable instanceof ConnectException) {
-
+					if (isCausedByConnectionException(throwable)) {
 						return hostProvider.getActive(Verification.ACTIVE) //
 								.flatMap(callback::doWithClient);
 					}
@@ -551,6 +550,27 @@ public <T> Mono<T> execute(ReactiveElasticsearchClientCallback<T> callback) {
 				});
 	}
 
+	/**
+	 * checks if the given throwable is a {@link ConnectException} or has one in it's cause chain
+	 *
+	 * @param throwable the throwable to check
+	 * @return true if throwable is caused by a {@link ConnectException}
+	 */
+	private boolean isCausedByConnectionException(Throwable throwable) {
+
+		Throwable t = throwable;
+		do {
+
+			if (t instanceof ConnectException) {
+				return true;
+			}
+
+			t = t.getCause();
+		} while (t != null);
+
+		return false;
+	}
+
 	@Override
 	public Mono<Status> status() {
 

src/main/java/org/springframework/data/elasticsearch/client/reactive/MultiNodeHostProvider.java

@@ -20,6 +20,7 @@
 import reactor.util.function.Tuple2;
 
 import java.net.InetSocketAddress;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -29,6 +30,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.data.elasticsearch.client.ElasticsearchHost;
 import org.springframework.data.elasticsearch.client.ElasticsearchHost.State;
 import org.springframework.data.elasticsearch.client.NoReachableHostException;
@@ -47,6 +50,8 @@
  */
 class MultiNodeHostProvider implements HostProvider<MultiNodeHostProvider> {
 
+	private final static Logger LOG = LoggerFactory.getLogger(MultiNodeHostProvider.class);
+
 	private final WebClientProvider clientProvider;
 	private final Supplier<HttpHeaders> headersSupplier;
 	private final Map<InetSocketAddress, ElasticsearchHost> hosts;
@@ -60,6 +65,8 @@ class MultiNodeHostProvider implements HostProvider<MultiNodeHostProvider> {
 		for (InetSocketAddress endpoint : endpoints) {
 			this.hosts.put(endpoint, new ElasticsearchHost(endpoint, State.UNKNOWN));
 		}
+
+		LOG.debug("initialized with " + hosts);
 	}
 
 	/*
@@ -68,7 +75,7 @@ class MultiNodeHostProvider implements HostProvider<MultiNodeHostProvider> {
 	 */
 	@Override
 	public Mono<ClusterInformation> clusterInfo() {
-		return nodes(null).map(this::updateNodeState).buffer(hosts.size())
+		return checkNodes(null).map(this::updateNodeState).buffer(hosts.size())
 				.then(Mono.just(new ClusterInformation(new LinkedHashSet<>(this.hosts.values()))));
 	}
 
@@ -88,14 +95,19 @@ public WebClient createWebClient(InetSocketAddress endpoint) {
 	@Override
 	public Mono<InetSocketAddress> lookupActiveHost(Verification verification) {
 
+		LOG.trace("lookupActiveHost " + verification + " from " + hosts());
+
 		if (Verification.LAZY.equals(verification)) {
 			for (ElasticsearchHost entry : hosts()) {
 				if (entry.isOnline()) {
+					LOG.trace("lookupActiveHost returning " + entry);
 					return Mono.just(entry.getEndpoint());
 				}
 			}
+			LOG.trace("no online host found with LAZY");
 		}
 
+		LOG.trace("searching for active host");
 		return findActiveHostInKnownActives() //
 				.switchIfEmpty(findActiveHostInUnresolved()) //
 				.switchIfEmpty(findActiveHostInDead()) //
@@ -107,20 +119,30 @@ Collection<ElasticsearchHost> getCachedHostState() {
 	}
 
 	private Mono<InetSocketAddress> findActiveHostInKnownActives() {
-		return findActiveForSate(State.ONLINE);
+		return findActiveForState(State.ONLINE);
 	}
 
 	private Mono<InetSocketAddress> findActiveHostInUnresolved() {
-		return findActiveForSate(State.UNKNOWN);
+		return findActiveForState(State.UNKNOWN);
 	}
 
 	private Mono<InetSocketAddress> findActiveHostInDead() {
-		return findActiveForSate(State.OFFLINE);
+		return findActiveForState(State.OFFLINE);
 	}
 
-	private Mono<InetSocketAddress> findActiveForSate(State state) {
-		return nodes(state).map(this::updateNodeState).filter(ElasticsearchHost::isOnline)
-				.map(ElasticsearchHost::getEndpoint).next();
+	private Mono<InetSocketAddress> findActiveForState(State state) {
+
+		LOG.trace("findActiveForState state " + state + ", current hosts: " + hosts);
+
+		return checkNodes(state) //
+				.map(this::updateNodeState) //
+				.filter(ElasticsearchHost::isOnline) //
+				.map(elasticsearchHost -> {
+					LOG.trace("findActiveForState returning host " + elasticsearchHost);
+					return elasticsearchHost;
+				}).map(ElasticsearchHost::getEndpoint) //
+				.takeLast(1) //
+				.next();
 	}
 
 	private ElasticsearchHost updateNodeState(Tuple2<InetSocketAddress, State> tuple2) {
@@ -131,28 +153,36 @@ private ElasticsearchHost updateNodeState(Tuple2<InetSocketAddress, State> tuple
 		return elasticsearchHost;
 	}
 
-	private Flux<Tuple2<InetSocketAddress, State>> nodes(@Nullable State state) {
+	private Flux<Tuple2<InetSocketAddress, State>> checkNodes(@Nullable State state) {
+
+		LOG.trace("checkNodes() with state " + state);
 
 		return Flux.fromIterable(hosts()) //
 				.filter(entry -> state == null || entry.getState().equals(state)) //
 				.map(ElasticsearchHost::getEndpoint) //
-				.flatMap(host -> {
+				.concatMap(host -> {
+
+					LOG.trace("checking host " + host);
 
 					Mono<ClientResponse> clientResponseMono = createWebClient(host) //
 							.head().uri("/") //
 							.headers(httpHeaders -> httpHeaders.addAll(headersSupplier.get())) //
 							.exchangeToMono(Mono::just) //
+							.timeout(Duration.ofSeconds(1)) //
 							.doOnError(throwable -> {
+								LOG.trace("error checking host " + host + ", " + throwable.getMessage());
 								hosts.put(host, new ElasticsearchHost(host, State.OFFLINE));
 								clientProvider.getErrorListener().accept(throwable);
 							});
 
 					return Mono.just(host) //
-							.zipWith( //
-									clientResponseMono.flatMap(it -> it.releaseBody() //
-											.thenReturn(it.statusCode().isError() ? State.OFFLINE : State.ONLINE)));
+							.zipWith(clientResponseMono.flatMap(it -> it.releaseBody() //
+									.thenReturn(it.statusCode().isError() ? State.OFFLINE : State.ONLINE)));
 				}) //
-				.onErrorContinue((throwable, o) -> clientProvider.getErrorListener().accept(throwable));
+				.map(tuple -> {
+					LOG.trace("check result " + tuple);
+					return tuple;
+				}).onErrorContinue((throwable, o) -> clientProvider.getErrorListener().accept(throwable));
 	}
 
 	private List<ElasticsearchHost> hosts() {