Cwikius-Spring/spring-data-elasticsearch
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-data-elasticsearch.git synced 2025-06-01 09:42:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix reactive connection handling.

Browse Source 
Original Pull Request #1766
Closes #1759
This commit is contained in:
Peter-Josef Meisch 2021-04-08 22:00:46 +02:00 committed by GitHub
parent 4782414596
commit 58bca88386
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 65 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -541,8 +541,7 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
.flatMap(callback::doWithClient) // .flatMap(callback::doWithClient) //
.onErrorResume(throwable -> { .onErrorResume(throwable -> {
if (throwable instanceof ConnectException) { if (isCausedByConnectionException(throwable)) {
return hostProvider.getActive(Verification.ACTIVE) // return hostProvider.getActive(Verification.ACTIVE) //
.flatMap(callback::doWithClient); .flatMap(callback::doWithClient);
} }
@ -551,6 +550,27 @@ public class DefaultReactiveElasticsearchClient implements ReactiveElasticsearch
}); });
} }
/**
* 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 @Override
public Mono<Status> status() { public Mono<Status> status() {

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

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