honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix concurrency bug when getting the host for a given request 

It can happen that the list of healthy hosts is empty, then we get one from the blacklist. but some other operation might have sneaked in and emptied the blacklist in the meantime, so we have to retry till we manage to get some host, either from the healthy list or from the blacklist.
This commit is contained in:
javanna 2016-07-22 19:03:45 +02:00 committed by Luca Cavanna
parent e6054a931e
commit 46cb3f36ff
1 changed files with 29 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
client/rest/src/main/java/org/elasticsearch/client/RestClient.java
View File

@ -46,6 +46,7 @@ import java.io.IOException;
import java.net.URI; import java.net.URI;
import java.net.URISyntaxException; import java.net.URISyntaxException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections; import java.util.Collections;
import java.util.Comparator; import java.util.Comparator;
import java.util.HashSet; import java.util.HashSet;
@ -373,16 +374,18 @@ public final class RestClient implements Closeable {
* In case there are no healthy hosts available, or dead ones to be be retried, one dead host gets returned. * In case there are no healthy hosts available, or dead ones to be be retried, one dead host gets returned.
*/ */
private Iterable<HttpHost> nextHost() { private Iterable<HttpHost> nextHost() {
Collection<HttpHost> nextHosts = Collections.emptySet();
do {
Set<HttpHost> filteredHosts = new HashSet<>(hosts); Set<HttpHost> filteredHosts = new HashSet<>(hosts);
for (Map.Entry<HttpHost, DeadHostState> entry : blacklist.entrySet()) { for (Map.Entry<HttpHost, DeadHostState> entry : blacklist.entrySet()) {
if (System.nanoTime() - entry.getValue().getDeadUntilNanos() < 0) { if (System.nanoTime() - entry.getValue().getDeadUntilNanos() < 0) {
filteredHosts.remove(entry.getKey()); filteredHosts.remove(entry.getKey());
} }
} }
if (filteredHosts.isEmpty()) { if (filteredHosts.isEmpty()) {
//last resort: if there are no good hosts to use, return a single dead one, the one that's closest to being retried //last resort: if there are no good hosts to use, return a single dead one, the one that's closest to being retried
List<Map.Entry<HttpHost, DeadHostState>> sortedHosts = new ArrayList<>(blacklist.entrySet()); List<Map.Entry<HttpHost, DeadHostState>> sortedHosts = new ArrayList<>(blacklist.entrySet());
if (sortedHosts.size() > 0) {
Collections.sort(sortedHosts, new Comparator<Map.Entry<HttpHost, DeadHostState>>() { Collections.sort(sortedHosts, new Comparator<Map.Entry<HttpHost, DeadHostState>>() {
@Override @Override
public int compare(Map.Entry<HttpHost, DeadHostState> o1, Map.Entry<HttpHost, DeadHostState> o2) { public int compare(Map.Entry<HttpHost, DeadHostState> o1, Map.Entry<HttpHost, DeadHostState> o2) {
@ -391,12 +394,15 @@ public final class RestClient implements Closeable {
}); });
HttpHost deadHost = sortedHosts.get(0).getKey(); HttpHost deadHost = sortedHosts.get(0).getKey();
logger.trace("resurrecting host [" + deadHost + "]"); logger.trace("resurrecting host [" + deadHost + "]");
return Collections.singleton(deadHost); nextHosts = Collections.singleton(deadHost);
} }
} else {
List<HttpHost> rotatedHosts = new ArrayList<>(filteredHosts); List<HttpHost> rotatedHosts = new ArrayList<>(filteredHosts);
Collections.rotate(rotatedHosts, rotatedHosts.size() - lastHostIndex.getAndIncrement()); Collections.rotate(rotatedHosts, rotatedHosts.size() - lastHostIndex.getAndIncrement());
return rotatedHosts; nextHosts = rotatedHosts;
}
} while(nextHosts.isEmpty());
return nextHosts;
} }
/** /**