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

Merge pull request #13349 from jasontedor/predicate-out-with-the-old-in-with-the-new 

Remove and forbid the use of com.google.common.base.Predicate(s)?
This commit is contained in:
Jason Tedor 2015-09-06 07:24:26 -04:00
parent e69551d8df 8a3411e5e4
commit 5e36c91e8c
57 changed files with 563 additions and 775 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
core/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/TransportGetFieldMappingsIndexAction.java
View File

@ -19,8 +19,6 @@
package org.elasticsearch.action.admin.indices.mapping.get;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableMap;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.admin.indices.mapping.get.GetFieldMappingsResponse.FieldMappingMetaData;
@ -51,8 +49,10 @@ import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportService;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.stream.Collectors;
import static org.elasticsearch.common.util.CollectionUtils.newLinkedList;
@ -96,14 +96,10 @@ public class TransportGetFieldMappingsIndexAction extends TransportSingleShardAc
typeIntersection = indexService.mapperService().types();
} else {
typeIntersection = Collections2.filter(indexService.mapperService().types(), new Predicate<String>() {
@Override
public boolean apply(String type) {
return Regex.simpleMatch(request.types(), type);
}
});
typeIntersection = indexService.mapperService().types()
.stream()
.filter(type -> Regex.simpleMatch(request.types(), type))
.collect(Collectors.toCollection(ArrayList::new));
if (typeIntersection.isEmpty()) {
throw new TypeMissingException(shardId.index(), request.types());
}

14
core/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cluster.metadata;
import com.google.common.base.Predicate;
import org.elasticsearch.ElasticsearchParseException;
import org.elasticsearch.action.IndicesRequest;
import org.elasticsearch.action.support.IndicesOptions;
@ -50,8 +49,8 @@ import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;
import static com.google.common.collect.Maps.filterEntries;
import static com.google.common.collect.Maps.newHashMap;
public class IndexNameExpressionResolver extends AbstractComponent {
@ -600,12 +599,11 @@ public class IndexNameExpressionResolver extends AbstractComponent {
} else {
// Other wildcard expressions:
final String pattern = expression;
matches = filterEntries(metaData.getAliasAndIndexLookup(), new Predicate<Map.Entry<String, AliasOrIndex>>() {
@Override
public boolean apply(@Nullable Map.Entry<String, AliasOrIndex> input) {
return Regex.simpleMatch(pattern, input.getKey());
}
});
matches = metaData.getAliasAndIndexLookup()
.entrySet()
.stream()
.filter(e -> Regex.simpleMatch(pattern, e.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
for (Map.Entry<String, AliasOrIndex> entry : matches.entrySet()) {
AliasOrIndex aliasOrIndex = entry.getValue();

38
core/src/main/java/org/elasticsearch/cluster/metadata/MetaData.java
View File

@ -22,8 +22,6 @@ package org.elasticsearch.cluster.metadata;
import com.carrotsearch.hppc.ObjectHashSet;
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.UnmodifiableIterator;
import org.apache.lucene.util.CollectionUtil;
@ -74,8 +72,11 @@ import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.stream.Collectors;
import static org.elasticsearch.common.settings.Settings.*;
import static org.elasticsearch.common.settings.Settings.readSettingsFromStream;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.common.settings.Settings.writeSettingsToStream;
public class MetaData implements Iterable<IndexMetaData>, Diffable<MetaData>, FromXContentBuilder<MetaData>, ToXContent {
@ -382,22 +383,23 @@ public class MetaData implements Iterable<IndexMetaData>, Diffable<MetaData>, Fr
}
// TODO: make this a List so we don't have to copy below
Collection<IndexWarmersMetaData.Entry> filteredWarmers = Collections2.filter(indexWarmersMetaData.entries(), new Predicate<IndexWarmersMetaData.Entry>() {
Collection<IndexWarmersMetaData.Entry> filteredWarmers =
indexWarmersMetaData
.entries()
.stream()
.filter(warmer -> {
if (warmers.length != 0 && types.length != 0) {
return Regex.simpleMatch(warmers, warmer.name()) && Regex.simpleMatch(types, warmer.types());
} else if (warmers.length != 0) {
return Regex.simpleMatch(warmers, warmer.name());
} else if (types.length != 0) {
return Regex.simpleMatch(types, warmer.types());
} else {
return true;
}
})
.collect(Collectors.toCollection(ArrayList::new));
@Override
public boolean apply(IndexWarmersMetaData.Entry warmer) {
if (warmers.length != 0 && types.length != 0) {
return Regex.simpleMatch(warmers, warmer.name()) && Regex.simpleMatch(types, warmer.types());
} else if (warmers.length != 0) {
return Regex.simpleMatch(warmers, warmer.name());
} else if (types.length != 0) {
return Regex.simpleMatch(types, warmer.types());
} else {
return true;
}
}
});
if (!filteredWarmers.isEmpty()) {
mapBuilder.put(index, Collections.unmodifiableList(new ArrayList<>(filteredWarmers)));
}

4
core/src/main/java/org/elasticsearch/cluster/routing/RoutingNodes.java
View File

@ -21,7 +21,6 @@ package org.elasticsearch.cluster.routing;
import com.carrotsearch.hppc.ObjectIntHashMap;
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import org.apache.lucene.util.CollectionUtil;
@ -40,6 +39,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
import static com.google.common.collect.Maps.newHashMap;
import static com.google.common.collect.Sets.newHashSet;
@ -294,7 +294,7 @@ public class RoutingNodes implements Iterable<RoutingNode> {
List<ShardRouting> shards = new ArrayList<>();
for (RoutingNode routingNode : this) {
for (ShardRouting shardRouting : routingNode) {
if (predicate.apply(shardRouting)) {
if (predicate.test(shardRouting)) {
shards.add(shardRouting);
}
}

26
core/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java
View File

@ -20,8 +20,6 @@
package org.elasticsearch.cluster.routing;
import com.carrotsearch.hppc.IntSet;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.UnmodifiableIterator;
@ -39,6 +37,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;
import static com.google.common.collect.Maps.newHashMap;
@ -184,19 +183,8 @@ public class RoutingTable implements Iterable<IndexRoutingTable>, Diffable<Routi
return allSatisfyingPredicateShardsGrouped(indices, includeEmpty, includeRelocationTargets, ASSIGNED_PREDICATE);
}
private static Predicate<ShardRouting> ACTIVE_PREDICATE = new Predicate<ShardRouting>() {
@Override
public boolean apply(ShardRouting shardRouting) {
return shardRouting.active();
}
};
private static Predicate<ShardRouting> ASSIGNED_PREDICATE = new Predicate<ShardRouting>() {
@Override
public boolean apply(ShardRouting shardRouting) {
return shardRouting.assignedToNode();
}
};
private static Predicate<ShardRouting> ACTIVE_PREDICATE = shardRouting -> shardRouting.active();
private static Predicate<ShardRouting> ASSIGNED_PREDICATE = shardRouting -> shardRouting.assignedToNode();
// TODO: replace with JDK 8 native java.util.function.Predicate
private GroupShardsIterator allSatisfyingPredicateShardsGrouped(String[] indices, boolean includeEmpty, boolean includeRelocationTargets, Predicate<ShardRouting> predicate) {
@ -210,7 +198,7 @@ public class RoutingTable implements Iterable<IndexRoutingTable>, Diffable<Routi
}
for (IndexShardRoutingTable indexShardRoutingTable : indexRoutingTable) {
for (ShardRouting shardRouting : indexShardRoutingTable) {
if (predicate.apply(shardRouting)) {
if (predicate.test(shardRouting)) {
set.add(shardRouting.shardsIt());
if (includeRelocationTargets && shardRouting.relocating()) {
set.add(new PlainShardIterator(shardRouting.shardId(), Collections.singletonList(shardRouting.buildTargetRelocatingShard())));
@ -225,11 +213,11 @@ public class RoutingTable implements Iterable<IndexRoutingTable>, Diffable<Routi
}
public ShardsIterator allShards(String[] indices) {
return allShardsSatisfyingPredicate(indices, Predicates.<ShardRouting>alwaysTrue(), false);
return allShardsSatisfyingPredicate(indices, shardRouting -> true, false);
}
public ShardsIterator allShardsIncludingRelocationTargets(String[] indices) {
return allShardsSatisfyingPredicate(indices, Predicates.<ShardRouting>alwaysTrue(), true);
return allShardsSatisfyingPredicate(indices, shardRouting -> true, true);
}
// TODO: replace with JDK 8 native java.util.function.Predicate
@ -244,7 +232,7 @@ public class RoutingTable implements Iterable<IndexRoutingTable>, Diffable<Routi
}
for (IndexShardRoutingTable indexShardRoutingTable : indexRoutingTable) {
for (ShardRouting shardRouting : indexShardRoutingTable) {
if (predicate.apply(shardRouting)) {
if (predicate.test(shardRouting)) {
shards.add(shardRouting);
if (includeRelocationTargets && shardRouting.relocating()) {
shards.add(shardRouting.buildTargetRelocatingShard());

10
core/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cluster.routing.allocation.allocator;
import com.google.common.base.Predicate;
import org.apache.lucene.util.ArrayUtil;
import org.apache.lucene.util.IntroSorter;
import org.elasticsearch.cluster.metadata.MetaData;
@ -40,6 +39,7 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.node.settings.NodeSettingsService;
import java.util.*;
import java.util.function.Predicate;
import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
@ -226,13 +226,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
private final float threshold;
private final MetaData metaData;
private final Predicate<ShardRouting> assignedFilter = new Predicate<ShardRouting>() {
@Override
public boolean apply(ShardRouting input) {
return input.assignedToNode();
}
};
private final Predicate<ShardRouting> assignedFilter = shard -> shard.assignedToNode();
public Balancer(ESLogger logger, RoutingAllocation allocation, WeightFunction weight, float threshold) {
this.logger = logger;

22
core/src/main/java/org/elasticsearch/common/settings/Settings.java
View File

@ -20,10 +20,8 @@
package org.elasticsearch.common.settings;
import com.google.common.base.Charsets;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSortedMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import org.elasticsearch.Version;
import org.elasticsearch.common.Booleans;
@ -1014,7 +1012,7 @@ public final class Settings implements ToXContent {
final Matcher matcher = ARRAY_PATTERN.matcher(entry.getKey());
if (matcher.matches()) {
prefixesToRemove.add(matcher.group(1));
} else if (Iterables.any(map.keySet(), startsWith(entry.getKey() + "."))) {
} else if (map.keySet().stream().anyMatch(key -> key.startsWith(entry.getKey() + "."))) {
prefixesToRemove.add(entry.getKey());
}
}
@ -1223,22 +1221,4 @@ public final class Settings implements ToXContent {
return new Settings(Collections.unmodifiableMap(map));
}
}
private static StartsWithPredicate startsWith(String prefix) {
return new StartsWithPredicate(prefix);
}
private static final class StartsWithPredicate implements Predicate<String> {
private String prefix;
public StartsWithPredicate(String prefix) {
this.prefix = prefix;
}
@Override
public boolean apply(String input) {
return input.startsWith(prefix);
}
}
}

13
core/src/main/java/org/elasticsearch/gateway/MetaDataStateFormat.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.gateway;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import org.apache.lucene.codecs.CodecUtil;
@ -48,9 +47,12 @@ import java.io.IOException;
import java.io.OutputStream;
import java.nio.file.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.function.Predicate;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
/**
* MetaDataStateFormat is a base class to write checksummed
@ -279,7 +281,12 @@ public abstract class MetaDataStateFormat<T> {
// new format (ie. legacy == false) then we know that the latest version state ought to use this new format.
// In case the state file with the latest version does not use the new format while older state files do,
// the list below will be empty and loading the state will fail
for (PathAndStateId pathAndStateId : Collections2.filter(files, new StateIdAndLegacyPredicate(maxStateId, maxStateIdIsLegacy))) {
Collection<PathAndStateId> pathAndStateIds = files
.stream()
.filter(new StateIdAndLegacyPredicate(maxStateId, maxStateIdIsLegacy))
.collect(Collectors.toCollection(ArrayList::new));
for (PathAndStateId pathAndStateId : pathAndStateIds) {
try {
final Path stateFile = pathAndStateId.file;
final long id = pathAndStateId.id;
@ -328,7 +335,7 @@ public abstract class MetaDataStateFormat<T> {
}
@Override
public boolean apply(PathAndStateId input) {
public boolean test(PathAndStateId input) {
return input.id == id && input.legacy == legacy;
}
}

10
core/src/main/java/org/elasticsearch/index/codec/postingsformat/Elasticsearch090PostingsFormat.java
View File

@ -18,8 +18,6 @@
*/
package org.elasticsearch.index.codec.postingsformat;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.Iterators;
import org.apache.lucene.codecs.FieldsConsumer;
import org.apache.lucene.codecs.FieldsProducer;
@ -36,6 +34,7 @@ import org.elasticsearch.index.mapper.internal.UidFieldMapper;
import java.io.IOException;
import java.util.Iterator;
import java.util.function.Predicate;
/**
* This is the old default postings format for Elasticsearch that special cases
@ -60,13 +59,8 @@ public class Elasticsearch090PostingsFormat extends PostingsFormat {
public PostingsFormat getDefaultWrapped() {
return bloomPostings.getDelegate();
}
protected static final Predicate<String> UID_FIELD_FILTER = new Predicate<String>() {
@Override
public boolean apply(String s) {
return UidFieldMapper.NAME.equals(s);
}
};
protected static final Predicate<String> UID_FIELD_FILTER = field -> UidFieldMapper.NAME.equals(field);
@Override
public FieldsConsumer fieldsConsumer(SegmentWriteState state) throws IOException {

11
core/src/main/java/org/elasticsearch/index/mapper/MapperService.java
View File

@ -21,7 +21,6 @@ package org.elasticsearch.index.mapper;
import com.carrotsearch.hppc.ObjectHashSet;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterators;
@ -71,6 +70,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Predicate;
import static org.elasticsearch.common.collect.MapBuilder.newMapBuilder;
@ -202,20 +202,13 @@ public class MapperService extends AbstractIndexComponent implements Closeable {
if (includingDefaultMapping) {
iterator = mappers.values().iterator();
} else {
iterator = Iterators.filter(mappers.values().iterator(), NOT_A_DEFAULT_DOC_MAPPER);
iterator = mappers.values().stream().filter(mapper -> !DEFAULT_MAPPING.equals(mapper.type())).iterator();
}
return Iterators.unmodifiableIterator(iterator);
}
};
}
private static final Predicate<DocumentMapper> NOT_A_DEFAULT_DOC_MAPPER = new Predicate<DocumentMapper>() {
@Override
public boolean apply(DocumentMapper input) {
return !DEFAULT_MAPPING.equals(input.type());
}
};
public AnalysisService analysisService() {
return this.analysisService;
}

9
core/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
View File

@ -21,7 +21,6 @@ package org.elasticsearch.indices.cluster;
import com.carrotsearch.hppc.IntHashSet;
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.google.common.base.Predicate;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
@ -524,13 +523,7 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent<Indic
} else if (isPeerRecovery(shardRouting)) {
final DiscoveryNode sourceNode = findSourceNodeForPeerRecovery(routingTable, nodes, shardRouting);
// check if there is an existing recovery going, and if so, and the source node is not the same, cancel the recovery to restart it
final Predicate<RecoveryStatus> shouldCancel = new Predicate<RecoveryStatus>() {
@Override
public boolean apply(@Nullable RecoveryStatus status) {
return status.sourceNode().equals(sourceNode) == false;
}
};
if (recoveryTarget.cancelRecoveriesForShard(indexShard.shardId(), "recovery source node changed", shouldCancel)) {
if (recoveryTarget.cancelRecoveriesForShard(indexShard.shardId(), "recovery source node changed", status -> !status.sourceNode().equals(sourceNode))) {
logger.debug("[{}][{}] removing shard (recovery source changed), current [{}], global [{}])", shardRouting.index(), shardRouting.id(), currentRoutingEntry, shardRouting);
// closing the shard will also cancel any ongoing recovery.
indexService.removeShard(shardRouting.id(), "removing shard (recovery source node changed)");

7
core/src/main/java/org/elasticsearch/indices/recovery/RecoveriesCollection.java
View File

@ -19,8 +19,6 @@
package org.elasticsearch.indices.recovery;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import org.elasticsearch.ElasticsearchTimeoutException;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.logging.ESLogger;
@ -34,6 +32,7 @@ import org.elasticsearch.threadpool.ThreadPool;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
/**
* This class holds a collection of all on going recoveries on the current node (i.e., the node is the target node
@ -139,7 +138,7 @@ public class RecoveriesCollection {
/** cancel all ongoing recoveries for the given shard. typically because the shards is closed */
public boolean cancelRecoveriesForShard(ShardId shardId, String reason) {
return cancelRecoveriesForShard(shardId, reason, Predicates.<RecoveryStatus>alwaysTrue());
return cancelRecoveriesForShard(shardId, reason, status -> true);
}
/**
@ -160,7 +159,7 @@ public class RecoveriesCollection {
// if we can't increment the status, the recovery is not there any more.
if (status.tryIncRef()) {
try {
cancel = shouldCancel.apply(status);
cancel = shouldCancel.test(status);
} finally {
status.decRef();
}

2
core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.indices.recovery;
import com.google.common.base.Predicate;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.IndexFormatTooNewException;
import org.apache.lucene.index.IndexFormatTooOldException;
@ -58,6 +57,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import static org.elasticsearch.common.unit.TimeValue.timeValueMillis;

11
core/src/main/java/org/elasticsearch/search/aggregations/LeafBucketCollector.java
View File

@ -19,13 +19,13 @@
package org.elasticsearch.search.aggregations;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import org.apache.lucene.search.LeafCollector;
import org.apache.lucene.search.Scorer;
import java.io.IOException;
import java.util.stream.StreamSupport;
/**
* Per-leaf bucket collector.
@ -44,13 +44,8 @@ public abstract class LeafBucketCollector implements LeafCollector {
};
public static LeafBucketCollector wrap(Iterable<LeafBucketCollector> collectors) {
final Iterable<LeafBucketCollector> actualCollectors = Iterables.filter(collectors,
new Predicate<LeafBucketCollector> () {
@Override
public boolean apply(LeafBucketCollector c) {
return c != NO_OP_COLLECTOR;
}
});
final Iterable<LeafBucketCollector> actualCollectors =
StreamSupport.stream(collectors.spliterator(), false).filter(c -> c != NO_OP_COLLECTOR)::iterator;
final LeafBucketCollector[] colls = Iterables.toArray(actualCollectors, LeafBucketCollector.class);
switch (colls.length) {
case 0:

4
core/src/test/java/org/elasticsearch/action/admin/indices/shards/IndicesShardStoreRequestIT.java
View File

@ -22,7 +22,6 @@ package org.elasticsearch.action.admin.indices.shards;
import com.carrotsearch.hppc.cursors.IntObjectCursor;
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.google.common.base.Predicate;
import org.apache.lucene.index.CorruptIndexException;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.client.Requests;
@ -42,6 +41,7 @@ import org.junit.Test;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.function.Predicate;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoTimeout;
@ -209,7 +209,7 @@ public class IndicesShardStoreRequestIT extends ESIntegTestCase {
}
@Override
public boolean apply(Settings settings) {
public boolean test(Settings settings) {
return nodesWithShard.contains(settings.get("name"));
}

22
core/src/test/java/org/elasticsearch/action/admin/indices/upgrade/UpgradeIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.action.admin.indices.upgrade;
import com.google.common.base.Predicate;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.Version;
import org.elasticsearch.action.admin.indices.segments.IndexSegments;
@ -43,6 +42,7 @@ import org.junit.BeforeClass;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.function.Predicate;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
@ -97,12 +97,7 @@ public class UpgradeIT extends ESBackcompatTestCase {
if (globalCompatibilityVersion().before(Version.V_1_4_0_Beta1)) {
// before 1.4 and the wait_if_ongoing flag, flushes could fail randomly, so we
// need to continue to try flushing until all shards succeed
assertTrue(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
return flush(indexName).getFailedShards() == 0;
}
}));
assertTrue(awaitBusy(() -> flush(indexName).getFailedShards() == 0));
} else {
assertEquals(0, flush(indexName).getFailedShards());
}
@ -143,14 +138,11 @@ public class UpgradeIT extends ESBackcompatTestCase {
logger.info("--> Running upgrade on index " + indexToUpgrade);
assertNoFailures(client().admin().indices().prepareUpgrade(indexToUpgrade).get());
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
try {
return isUpgraded(client(), indexToUpgrade);
} catch (Exception e) {
throw ExceptionsHelper.convertToRuntime(e);
}
awaitBusy(() -> {
try {
return isUpgraded(client(), indexToUpgrade);
} catch (Exception e) {
throw ExceptionsHelper.convertToRuntime(e);
}
});
logger.info("--> Single index upgrade complete");

34
core/src/test/java/org/elasticsearch/action/support/replication/ShardReplicationTests.java
View File

@ -18,17 +18,14 @@
*/
package org.elasticsearch.action.support.replication;
import com.google.common.base.Predicate;
import org.apache.lucene.index.CorruptIndexException;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionWriteResponse;
import org.elasticsearch.action.UnavailableShardsException;
import org.elasticsearch.action.WriteConsistencyLevel;
import org.elasticsearch.action.support.ActionFilter;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.PlainActionFuture;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateObserver;
@ -40,16 +37,15 @@ import org.elasticsearch.cluster.block.ClusterBlocks;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.*;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.ShardIterator;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.lease.Releasable;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.DummyTransportAddress;
import org.elasticsearch.index.shard.IndexShardNotStartedException;
import org.elasticsearch.index.shard.IndexShardState;
import org.elasticsearch.index.shard.ShardId;
@ -70,7 +66,6 @@ import org.junit.Test;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@ -79,8 +74,11 @@ import java.util.concurrent.atomic.AtomicInteger;
import static org.elasticsearch.action.support.replication.ClusterStateCreationUtils.state;
import static org.elasticsearch.action.support.replication.ClusterStateCreationUtils.stateWithStartedPrimary;
import static org.elasticsearch.cluster.metadata.IndexMetaData.*;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.arrayWithSize;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
public class ShardReplicationTests extends ESTestCase {
@ -441,12 +439,7 @@ public class ShardReplicationTests extends ESTestCase {
t.start();
// shard operation should be ongoing, so the counter is at 2
// we have to wait here because increment happens in thread
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(@Nullable Object input) {
return (count.get() == 2);
}
});
awaitBusy(() -> count.get() == 2);
assertIndexShardCounter(2);
assertThat(transport.capturedRequests().length, equalTo(0));
@ -505,12 +498,7 @@ public class ShardReplicationTests extends ESTestCase {
t.start();
// shard operation should be ongoing, so the counter is at 2
// we have to wait here because increment happens in thread
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(@Nullable Object input) {
return count.get() == 2;
}
});
awaitBusy(() -> count.get() == 2);
((ActionWithDelay) action).countDownLatch.countDown();
t.join();
// operation should have finished and counter decreased because no outstanding replica requests

14
core/src/test/java/org/elasticsearch/bwcompat/OldIndexBackwardsCompatibilityIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.bwcompat;
import com.google.common.base.Predicate;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.lucene.index.IndexWriter;
@ -292,13 +291,12 @@ public class OldIndexBackwardsCompatibilityIT extends ESIntegTestCase {
* Waits for the index to show up in the cluster state in closed state
*/
void ensureClosed(final String index) throws InterruptedException {
assertTrue(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
ClusterState state = client().admin().cluster().prepareState().get().getState();
return state.metaData().hasIndex(index) && state.metaData().index(index).getState() == IndexMetaData.State.CLOSE;
}
}));
assertTrue(awaitBusy(() -> {
ClusterState state = client().admin().cluster().prepareState().get().getState();
return state.metaData().hasIndex(index) && state.metaData().index(index).getState() == IndexMetaData.State.CLOSE;
}
)
);
}
/**

9
core/src/test/java/org/elasticsearch/cache/recycler/MockPageCacheRecycler.java vendored
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cache.recycler;
import com.google.common.base.Predicate;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.elasticsearch.common.inject.Inject;
@ -45,12 +44,8 @@ public class MockPageCacheRecycler extends PageCacheRecycler {
// not empty, we might be executing on a shared cluster that keeps on obtaining
// and releasing pages, lets make sure that after a reasonable timeout, all master
// copy (snapshot) have been released
boolean success = ESTestCase.awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return Sets.intersection(masterCopy.keySet(), ACQUIRED_PAGES.keySet()).isEmpty();
}
});
boolean success =
ESTestCase.awaitBusy(() -> Sets.intersection(masterCopy.keySet(), ACQUIRED_PAGES.keySet()).isEmpty());
if (!success) {
masterCopy.keySet().retainAll(ACQUIRED_PAGES.keySet());
ACQUIRED_PAGES.keySet().removeAll(masterCopy.keySet()); // remove all existing master copy we will report on

9
core/src/test/java/org/elasticsearch/client/transport/TransportClientRetryIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.client.transport;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.support.PlainListenableActionFuture;
@ -30,7 +29,6 @@ import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.node.internal.InternalSettingsPreparer;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.InternalTestCluster;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.elasticsearch.transport.TransportService;
import org.junit.Test;
@ -72,12 +70,7 @@ public class TransportClientRetryIT extends ESIntegTestCase {
int size = cluster().size();
//kill all nodes one by one, leaving a single master/data node at the end of the loop
for (int j = 1; j < size; j++) {
internalCluster().stopRandomNode(new Predicate<Settings>() {
@Override
public boolean apply(Settings input) {
return true;
}
});
internalCluster().stopRandomNode(input -> true);
ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest().local(true);
ClusterState clusterState;

15
core/src/test/java/org/elasticsearch/cluster/ClusterModuleTests.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cluster;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.indices.create.CreateIndexClusterStateUpdateRequest;
import org.elasticsearch.cluster.metadata.IndexTemplateFilter;
import org.elasticsearch.cluster.metadata.IndexTemplateMetaData;
@ -85,12 +84,7 @@ public class ClusterModuleTests extends ModuleTestCase {
public void testRegisterClusterDynamicSetting() {
ClusterModule module = new ClusterModule(Settings.EMPTY);
module.registerClusterDynamicSetting("foo.bar", Validator.EMPTY);
assertInstanceBindingWithAnnotation(module, DynamicSettings.class, new Predicate<DynamicSettings>() {
@Override
public boolean apply(DynamicSettings dynamicSettings) {
return dynamicSettings.hasDynamicSetting("foo.bar");
}
}, ClusterDynamicSettings.class);
assertInstanceBindingWithAnnotation(module, DynamicSettings.class, dynamicSettings -> dynamicSettings.hasDynamicSetting("foo.bar"), ClusterDynamicSettings.class);
}
public void testRegisterIndexDynamicSettingDuplicate() {
@ -105,12 +99,7 @@ public class ClusterModuleTests extends ModuleTestCase {
public void testRegisterIndexDynamicSetting() {
ClusterModule module = new ClusterModule(Settings.EMPTY);
module.registerIndexDynamicSetting("foo.bar", Validator.EMPTY);
assertInstanceBindingWithAnnotation(module, DynamicSettings.class, new Predicate<DynamicSettings>() {
@Override
public boolean apply(DynamicSettings dynamicSettings) {
return dynamicSettings.hasDynamicSetting("foo.bar");
}
}, IndexDynamicSettings.class);
assertInstanceBindingWithAnnotation(module, DynamicSettings.class, dynamicSettings -> dynamicSettings.hasDynamicSetting("foo.bar"), IndexDynamicSettings.class);
}
public void testRegisterAllocationDeciderDuplicate() {

15
core/src/test/java/org/elasticsearch/cluster/ClusterServiceIT.java
View File

@ -18,7 +18,6 @@
*/
package org.elasticsearch.cluster;
import com.google.common.base.Predicate;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
@ -570,12 +569,7 @@ public class ClusterServiceIT extends ESIntegTestCase {
invoked2.await();
// whenever we test for no tasks, we need to awaitBusy since this is a live node
assertTrue(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return clusterService.pendingTasks().isEmpty();
}
}));
assertTrue(awaitBusy(() -> clusterService.pendingTasks().isEmpty()));
waitNoPendingTasksOnAll();
final CountDownLatch block2 = new CountDownLatch(1);
@ -688,12 +682,7 @@ public class ClusterServiceIT extends ESIntegTestCase {
internalCluster().stopRandomNonMasterNode();
// there should not be any master as the minimum number of required eligible masters is not met
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object obj) {
return clusterService1.state().nodes().masterNode() == null && clusterService1.state().status() == ClusterState.ClusterStateStatus.APPLIED;
}
});
awaitBusy(() -> clusterService1.state().nodes().masterNode() == null && clusterService1.state().status() == ClusterState.ClusterStateStatus.APPLIED);
assertThat(testService1.master(), is(false));
// bring the node back up

51
core/src/test/java/org/elasticsearch/cluster/MinimumMasterNodesIT.java
View File

@ -19,8 +19,6 @@
package org.elasticsearch.cluster;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.Priority;
@ -36,12 +34,16 @@ import org.junit.Test;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.test.ESIntegTestCase.Scope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.isOneOf;
import static org.hamcrest.Matchers.not;
@ClusterScope(scope = Scope.TEST, numDataNodes = 0)
@ESIntegTestCase.SuppressLocalMode
@ -98,12 +100,9 @@ public class MinimumMasterNodesIT extends ESIntegTestCase {
}
internalCluster().stopCurrentMasterNode();
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object obj) {
ClusterState state = client().admin().cluster().prepareState().setLocal(true).execute().actionGet().getState();
return state.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID);
}
awaitBusy(() -> {
ClusterState clusterState = client().admin().cluster().prepareState().setLocal(true).execute().actionGet().getState();
return clusterState.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID);
});
state = client().admin().cluster().prepareState().setLocal(true).execute().actionGet().getState();
assertThat(state.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID), equalTo(true));
@ -278,20 +277,26 @@ public class MinimumMasterNodesIT extends ESIntegTestCase {
}
private void assertNoMasterBlockOnAllNodes() throws InterruptedException {
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object obj) {
boolean success = true;
for (Client client : internalCluster()) {
ClusterState state = client.admin().cluster().prepareState().setLocal(true).execute().actionGet().getState();
success &= state.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID);
if (logger.isDebugEnabled()) {
logger.debug("Checking for NO_MASTER_BLOCK on client: {} NO_MASTER_BLOCK: [{}]", client, state.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID));
}
}
return success;
}
}, 20, TimeUnit.SECONDS), equalTo(true));
Predicate<Client> hasNoMasterBlock = client -> {
ClusterState state = client.admin().cluster().prepareState().setLocal(true).execute().actionGet().getState();
return state.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID);
};
assertTrue(awaitBusy(
() -> {
boolean success = true;
for (Client client : internalCluster()) {
boolean clientHasNoMasterBlock = hasNoMasterBlock.test(client);
if (logger.isDebugEnabled()) {
logger.debug("Checking for NO_MASTER_BLOCK on client: {} NO_MASTER_BLOCK: [{}]", client, clientHasNoMasterBlock);
}
success &= clientHasNoMasterBlock;
}
return success;
},
20,
TimeUnit.SECONDS
)
);
}
@Test

15
core/src/test/java/org/elasticsearch/cluster/NoMasterNodeIT.java
View File

@ -19,8 +19,6 @@
package org.elasticsearch.cluster;
import com.google.common.base.Predicate;
import org.elasticsearch.action.ActionRequestBuilder;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
@ -244,14 +242,11 @@ public class NoMasterNodeIT extends ESIntegTestCase {
logger.info("Cluster state:\n" + clusterState.getState().prettyPrint());
internalCluster().stopRandomDataNode();
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
ClusterState state = client().admin().cluster().prepareState().setLocal(true).get().getState();
return state.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID);
}
}), equalTo(true));
assertTrue(awaitBusy(() -> {
ClusterState state = client().admin().cluster().prepareState().setLocal(true).get().getState();
return state.blocks().hasGlobalBlock(DiscoverySettings.NO_MASTER_BLOCK_ID);
}
));
GetResponse getResponse = client().prepareGet("test1", "type1", "1").get();
assertExists(getResponse);

49
core/src/test/java/org/elasticsearch/cluster/ack/AckIT.java
View File

@ -19,6 +19,7 @@
package org.elasticsearch.cluster.ack;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesResponse;
@ -34,8 +35,8 @@ import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.AliasOrIndex;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
import org.elasticsearch.common.settings.Settings;
@ -44,17 +45,19 @@ import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.warmer.IndexWarmersMetaData;
import org.elasticsearch.test.ESIntegTestCase;
import org.junit.Test;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import com.google.common.base.Predicate;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.elasticsearch.cluster.metadata.IndexMetaData.*;
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS;
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
import static org.elasticsearch.cluster.metadata.IndexMetaData.State;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.notNullValue;
@ClusterScope(minNumDataNodes = 2)
public class AckIT extends ESIntegTestCase {
@ -120,17 +123,14 @@ public class AckIT extends ESIntegTestCase {
/* Since we don't wait for the ack here we have to wait until the search request has been executed from the master
* otherwise the test infra might have already deleted the index and the search request fails on all shards causing
* the test to fail too. We simply wait until the the warmer has been installed and also clean it up afterwards.*/
assertTrue(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
for (Client client : clients()) {
GetWarmersResponse getWarmersResponse = client.admin().indices().prepareGetWarmers().setLocal(true).get();
if (getWarmersResponse.warmers().size() != 1) {
return false;
}
assertTrue(awaitBusy(() -> {
for (Client client : clients()) {
GetWarmersResponse getWarmersResponse = client.admin().indices().prepareGetWarmers().setLocal(true).get();
if (getWarmersResponse.warmers().size() != 1) {
return false;
}
return true;
}
return true;
}));
assertAcked(client().admin().indices().prepareDeleteWarmer().setIndices("test").setNames("custom_warmer"));
}
@ -161,26 +161,23 @@ public class AckIT extends ESIntegTestCase {
DeleteWarmerResponse deleteWarmerResponse = client().admin().indices().prepareDeleteWarmer().setIndices("test").setNames("custom_warmer").setTimeout("0s").get();
assertFalse(deleteWarmerResponse.isAcknowledged());
assertTrue(awaitBusy(new Predicate<Object>() { // wait until they are all deleted
@Override
public boolean apply(Object input) {
for (Client client : clients()) {
GetWarmersResponse getWarmersResponse = client.admin().indices().prepareGetWarmers().setLocal(true).get();
if (getWarmersResponse.warmers().size() > 0) {
return false;
}
assertTrue(awaitBusy(() -> {
for (Client client : clients()) {
GetWarmersResponse getWarmersResponse = client.admin().indices().prepareGetWarmers().setLocal(true).get();
if (getWarmersResponse.warmers().size() > 0) {
return false;
}
return true;
}
return true;
}));
}
@Test
public void testClusterRerouteAcknowledgement() throws InterruptedException {
assertAcked(prepareCreate("test").setSettings(Settings.builder()
.put(indexSettings())
.put(SETTING_NUMBER_OF_SHARDS, between(cluster().numDataNodes(), DEFAULT_MAX_NUM_SHARDS))
.put(SETTING_NUMBER_OF_REPLICAS, 0)
.put(indexSettings())
.put(SETTING_NUMBER_OF_SHARDS, between(cluster().numDataNodes(), DEFAULT_MAX_NUM_SHARDS))
.put(SETTING_NUMBER_OF_REPLICAS, 0)
));
ensureGreen();

43
core/src/test/java/org/elasticsearch/cluster/allocation/AwarenessAllocationIT.java
View File

@ -20,7 +20,6 @@
package org.elasticsearch.cluster.allocation;
import com.carrotsearch.hppc.ObjectIntHashMap;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
@ -78,30 +77,30 @@ public class AwarenessAllocationIT extends ESIntegTestCase {
final String node3 = internalCluster().startNode(Settings.settingsBuilder().put(commonSettings).put("node.rack_id", "rack_2").build());
// On slow machines the initial relocation might be delayed
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
assertThat(awaitBusy(
() -> {
logger.info("--> waiting for no relocation");
ClusterHealthResponse clusterHealth = client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().setWaitForNodes("3").setWaitForRelocatingShards(0).get();
if (clusterHealth.isTimedOut()) {
return false;
}
logger.info("--> waiting for no relocation");
ClusterHealthResponse clusterHealth = client().admin().cluster().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().setWaitForNodes("3").setWaitForRelocatingShards(0).get();
if (clusterHealth.isTimedOut()) {
return false;
}
logger.info("--> checking current state");
ClusterState clusterState = client().admin().cluster().prepareState().execute().actionGet().getState();
// verify that we have all the primaries on node3
ObjectIntHashMap<String> counts = new ObjectIntHashMap<>();
for (IndexRoutingTable indexRoutingTable : clusterState.routingTable()) {
for (IndexShardRoutingTable indexShardRoutingTable : indexRoutingTable) {
for (ShardRouting shardRouting : indexShardRoutingTable) {
counts.addTo(clusterState.nodes().get(shardRouting.currentNodeId()).name(), 1);
logger.info("--> checking current state");
ClusterState clusterState = client().admin().cluster().prepareState().execute().actionGet().getState();
// verify that we have all the primaries on node3
ObjectIntHashMap<String> counts = new ObjectIntHashMap<>();
for (IndexRoutingTable indexRoutingTable : clusterState.routingTable()) {
for (IndexShardRoutingTable indexShardRoutingTable : indexRoutingTable) {
for (ShardRouting shardRouting : indexShardRoutingTable) {
counts.addTo(clusterState.nodes().get(shardRouting.currentNodeId()).name(), 1);
}
}
}
}
return counts.get(node3) == totalPrimaries;
}
}, 10, TimeUnit.SECONDS), equalTo(true));
return counts.get(node3) == totalPrimaries;
},
10,
TimeUnit.SECONDS
), equalTo(true));
}
@Test

34
core/src/test/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDeciderTests.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cluster.routing.allocation.decider;
import com.google.common.base.Predicate;
import org.elasticsearch.Version;
import org.elasticsearch.cluster.ClusterInfo;
import org.elasticsearch.cluster.ClusterInfoService;
@ -29,7 +28,14 @@ import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.*;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.RoutingNodes;
import org.elasticsearch.cluster.routing.RoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.cluster.routing.TestShardRouting;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocators;
@ -37,15 +43,24 @@ import org.elasticsearch.cluster.routing.allocation.command.AllocationCommand;
import org.elasticsearch.cluster.routing.allocation.command.AllocationCommands;
import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.test.ESAllocationTestCase;
import org.elasticsearch.test.gateway.NoopGatewayAllocator;
import org.elasticsearch.common.transport.LocalTransportAddress;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.test.ESAllocationTestCase;
import org.elasticsearch.test.gateway.NoopGatewayAllocator;
import org.junit.Test;
import java.util.*;
import java.util.AbstractMap;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
import static org.elasticsearch.cluster.routing.ShardRoutingState.INITIALIZING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.STARTED;
import static org.elasticsearch.cluster.routing.ShardRoutingState.UNASSIGNED;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
@ -899,12 +914,7 @@ public class DiskThresholdDeciderTests extends ESAllocationTestCase {
public void logShardStates(ClusterState state) {
RoutingNodes rn = state.getRoutingNodes();
logger.info("--> counts: total: {}, unassigned: {}, initializing: {}, relocating: {}, started: {}",
rn.shards(new Predicate<ShardRouting>() {
@Override
public boolean apply(ShardRouting input) {
return true;
}
}).size(),
rn.shards(shard -> true).size(),
rn.shardsWithState(UNASSIGNED).size(),
rn.shardsWithState(INITIALIZING).size(),
rn.shardsWithState(RELOCATING).size(),

4
core/src/test/java/org/elasticsearch/common/inject/ModuleTestCase.java
View File

@ -18,7 +18,6 @@
*/
package org.elasticsearch.common.inject;
import com.google.common.base.Predicate;
import org.elasticsearch.common.inject.spi.Element;
import org.elasticsearch.common.inject.spi.Elements;
import org.elasticsearch.common.inject.spi.InstanceBinding;
@ -34,6 +33,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
/**
* Base testcase for testing {@link Module} implementations.
@ -160,7 +160,7 @@ public abstract class ModuleTestCase extends ESTestCase {
InstanceBinding binding = (InstanceBinding) element;
if (to.equals(binding.getKey().getTypeLiteral().getType())) {
if (annotation == null || annotation.equals(binding.getKey().getAnnotationType())) {
assertTrue(tester.apply(to.cast(binding.getInstance())));
assertTrue(tester.test(to.cast(binding.getInstance())));
return;
}
}

8
core/src/test/java/org/elasticsearch/common/util/MockBigArrays.java
View File

@ -21,7 +21,6 @@ package org.elasticsearch.common.util;
import com.carrotsearch.randomizedtesting.RandomizedContext;
import com.carrotsearch.randomizedtesting.SeedUtils;
import com.google.common.base.Predicate;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.lucene.util.Accountable;
@ -59,12 +58,7 @@ public class MockBigArrays extends BigArrays {
// not empty, we might be executing on a shared cluster that keeps on obtaining
// and releasing arrays, lets make sure that after a reasonable timeout, all master
// copy (snapshot) have been released
boolean success = ESTestCase.awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return Sets.intersection(masterCopy.keySet(), ACQUIRED_ARRAYS.keySet()).isEmpty();
}
});
boolean success = ESTestCase.awaitBusy(() -> Sets.intersection(masterCopy.keySet(), ACQUIRED_ARRAYS.keySet()).isEmpty());
if (!success) {
masterCopy.keySet().retainAll(ACQUIRED_ARRAYS.keySet());
ACQUIRED_ARRAYS.keySet().removeAll(masterCopy.keySet()); // remove all existing master copy we will report on

56
core/src/test/java/org/elasticsearch/discovery/DiscoveryWithServiceDisruptionsIT.java
View File

@ -19,9 +19,7 @@
package org.elasticsearch.discovery;
import com.google.common.base.Predicate;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.lucene.util.LuceneTestCase.AwaitsFix;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
@ -32,7 +30,11 @@ import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.*;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.ClusterStateUpdateTask;
import org.elasticsearch.cluster.action.shard.ShardStateAction;
import org.elasticsearch.cluster.block.ClusterBlock;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
@ -65,7 +67,16 @@ import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.InternalTestCluster;
import org.elasticsearch.test.discovery.ClusterDiscoveryConfiguration;
import org.elasticsearch.test.disruption.*;
import org.elasticsearch.test.disruption.BlockClusterStateProcessing;
import org.elasticsearch.test.disruption.IntermittentLongGCDisruption;
import org.elasticsearch.test.disruption.LongGCDisruption;
import org.elasticsearch.test.disruption.NetworkDelaysPartition;
import org.elasticsearch.test.disruption.NetworkDisconnectPartition;
import org.elasticsearch.test.disruption.NetworkPartition;
import org.elasticsearch.test.disruption.NetworkUnresponsivePartition;
import org.elasticsearch.test.disruption.ServiceDisruptionScheme;
import org.elasticsearch.test.disruption.SingleNodeDisruption;
import org.elasticsearch.test.disruption.SlowClusterStateProcessing;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.elasticsearch.test.transport.MockTransportService;
import org.elasticsearch.transport.TransportException;
@ -76,8 +87,22 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@ -86,7 +111,10 @@ import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import static org.elasticsearch.test.ESIntegTestCase.Scope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.nullValue;
@ClusterScope(scope = Scope.TEST, numDataNodes = 0, transportClientRatio = 0)
@ESIntegTestCase.SuppressLocalMode
@ -1354,12 +1382,14 @@ public class DiscoveryWithServiceDisruptionsIT extends ESIntegTestCase {
private void assertDiscoveryCompleted(List<String> nodes) throws InterruptedException {
for (final String node : nodes) {
assertTrue("node [" + node + "] is still joining master", awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return !((ZenDiscovery) internalCluster().getInstance(Discovery.class, node)).joiningCluster();
}
}, 30, TimeUnit.SECONDS));
assertTrue(
"node [" + node + "] is still joining master",
awaitBusy(
() -> !((ZenDiscovery) internalCluster().getInstance(Discovery.class, node)).joiningCluster(),
30,
TimeUnit.SECONDS
)
);
}
}
}

8
core/src/test/java/org/elasticsearch/gateway/AsyncShardFetchTests.java
View File

@ -18,7 +18,6 @@
*/
package org.elasticsearch.gateway;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableSet;
import org.elasticsearch.Version;
import org.elasticsearch.action.FailedNodeException;
@ -286,12 +285,7 @@ public class AsyncShardFetchTests extends ESTestCase {
entry = simulations.get(nodeId);
if (entry == null) {
// we are simulating a master node switch, wait for it to not be null
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return simulations.containsKey(nodeId);
}
});
awaitBusy(() -> simulations.containsKey(nodeId));
}
assert entry != null;
entry.executeLatch.await();

22
core/src/test/java/org/elasticsearch/gateway/MetaDataWriteDataNodesIT.java
View File

@ -20,7 +20,6 @@
package org.elasticsearch.gateway;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.cluster.metadata.IndexMetaData;
@ -160,18 +159,15 @@ public class MetaDataWriteDataNodesIT extends ESIntegTestCase {
private void assertMetaState(final String nodeName, final String indexName, final boolean shouldBe) throws Exception {
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
logger.info("checking if meta state exists...");
try {
return shouldBe == metaStateExists(nodeName, indexName);
} catch (Throwable t) {
logger.info("failed to load meta state", t);
// TODO: loading of meta state fails rarely if the state is deleted while we try to load it
// this here is a hack, would be much better to use for example a WatchService
return false;
}
awaitBusy(() -> {
logger.info("checking if meta state exists...");
try {
return shouldBe == metaStateExists(nodeName, indexName);
} catch (Throwable t) {
logger.info("failed to load meta state", t);
// TODO: loading of meta state fails rarely if the state is deleted while we try to load it
// this here is a hack, would be much better to use for example a WatchService
return false;
}
});
boolean inMetaSate = metaStateExists(nodeName, indexName);

32
core/src/test/java/org/elasticsearch/gateway/QuorumGatewayIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.gateway;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
@ -36,10 +35,12 @@ import static org.elasticsearch.client.Requests.clusterHealthRequest;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
import static org.elasticsearch.test.ESIntegTestCase.*;
import static org.elasticsearch.test.ESIntegTestCase.Scope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
/**
*
@ -92,12 +93,10 @@ public class QuorumGatewayIT extends ESIntegTestCase {
ClusterHealthResponse clusterHealth = client().admin().cluster().health(clusterHealthRequest().waitForNodes("1")).actionGet();
assertThat(clusterHealth.isTimedOut(), equalTo(false));
assertThat(clusterHealth.getStatus(), equalTo(ClusterHealthStatus.RED)); // nothing allocated yet
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
ClusterStateResponse clusterStateResponse = internalCluster().smartClient().admin().cluster().prepareState().setMasterNodeTimeout("500ms").get();
return clusterStateResponse.getState() != null && clusterStateResponse.getState().routingTable().index("test") != null;
}}), equalTo(true)); // wait until we get a cluster state - could be null if we quick enough.
assertTrue(awaitBusy(() -> {
ClusterStateResponse clusterStateResponse = internalCluster().smartClient().admin().cluster().prepareState().setMasterNodeTimeout("500ms").get();
return clusterStateResponse.getState() != null && clusterStateResponse.getState().routingTable().index("test") != null;
})); // wait until we get a cluster state - could be null if we quick enough.
final ClusterStateResponse clusterStateResponse = internalCluster().smartClient().admin().cluster().prepareState().setMasterNodeTimeout("500ms").get();
assertThat(clusterStateResponse.getState(), notNullValue());
assertThat(clusterStateResponse.getState().routingTable().index("test"), notNullValue());
@ -149,15 +148,12 @@ public class QuorumGatewayIT extends ESIntegTestCase {
@Override
public void doAfterNodes(int numNodes, final Client activeClient) throws Exception {
if (numNodes == 1) {
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
logger.info("--> running cluster_health (wait for the shards to startup)");
ClusterHealthResponse clusterHealth = activeClient.admin().cluster().health(clusterHealthRequest().waitForYellowStatus().waitForNodes("2").waitForActiveShards(test.numPrimaries * 2)).actionGet();
logger.info("--> done cluster_health, status " + clusterHealth.getStatus());
return (!clusterHealth.isTimedOut()) && clusterHealth.getStatus() == ClusterHealthStatus.YELLOW;
}
}, 30, TimeUnit.SECONDS), equalTo(true));
assertTrue(awaitBusy(() -> {
logger.info("--> running cluster_health (wait for the shards to startup)");
ClusterHealthResponse clusterHealth = activeClient.admin().cluster().health(clusterHealthRequest().waitForYellowStatus().waitForNodes("2").waitForActiveShards(test.numPrimaries * 2)).actionGet();
logger.info("--> done cluster_health, status " + clusterHealth.getStatus());
return (!clusterHealth.isTimedOut()) && clusterHealth.getStatus() == ClusterHealthStatus.YELLOW;
}, 30, TimeUnit.SECONDS));
logger.info("--> one node is closed -- index 1 document into the remaining nodes");
activeClient.prepareIndex("test", "type1", "3").setSource(jsonBuilder().startObject().field("field", "value3").endObject()).get();
assertNoFailures(activeClient.admin().indices().prepareRefresh().get());

15
core/src/test/java/org/elasticsearch/index/TransportIndexFailuresIT.java
View File

@ -19,14 +19,12 @@
package org.elasticsearch.index;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableSet;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.index.IndexAction;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.RoutingNodes;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
@ -36,14 +34,16 @@ import org.elasticsearch.discovery.zen.fd.FaultDetection;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.transport.MockTransportService;
import org.elasticsearch.transport.TransportModule;
import org.elasticsearch.transport.TransportService;
import org.junit.Test;
import java.util.Collection;
import java.util.List;
import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
import static org.elasticsearch.cluster.routing.ShardRoutingState.INITIALIZING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.STARTED;
import static org.elasticsearch.cluster.routing.ShardRoutingState.UNASSIGNED;
import static org.hamcrest.Matchers.equalTo;
/**
@ -142,12 +142,7 @@ public class TransportIndexFailuresIT extends ESIntegTestCase {
state = getNodeClusterState(randomFrom(nodes.toArray(Strings.EMPTY_ARRAY)));
RoutingNodes rn = state.getRoutingNodes();
logger.info("--> counts: total: {}, unassigned: {}, initializing: {}, relocating: {}, started: {}",
rn.shards(new Predicate<ShardRouting>() {
@Override
public boolean apply(ShardRouting input) {
return true;
}
}).size(),
rn.shards(input -> true).size(),
rn.shardsWithState(UNASSIGNED).size(),
rn.shardsWithState(INITIALIZING).size(),
rn.shardsWithState(RELOCATING).size(),

7
core/src/test/java/org/elasticsearch/index/codec/postingformat/Elasticsearch090RWPostingsFormat.java
View File

@ -19,22 +19,21 @@
package org.elasticsearch.index.codec.postingformat;
import com.google.common.base.Predicates;
import com.google.common.collect.Iterators;
import org.apache.lucene.codecs.FieldsConsumer;
import org.apache.lucene.codecs.PostingsFormat;
import org.apache.lucene.index.Fields;
import org.apache.lucene.index.FilterLeafReader;
import org.apache.lucene.index.SegmentWriteState;
import org.elasticsearch.common.util.BloomFilter;
import org.elasticsearch.index.codec.postingsformat.BloomFilterPostingsFormat.BloomFilteredFieldsConsumer;
import org.elasticsearch.index.codec.postingsformat.BloomFilterPostingsFormat;
import org.elasticsearch.index.codec.postingsformat.BloomFilterPostingsFormat.BloomFilteredFieldsConsumer;
import org.elasticsearch.index.codec.postingsformat.Elasticsearch090PostingsFormat;
import org.elasticsearch.index.mapper.internal.UidFieldMapper;
import java.io.IOException;
import java.util.Iterator;
import java.util.stream.StreamSupport;
/** read-write version with blooms for testing */
public class Elasticsearch090RWPostingsFormat extends Elasticsearch090PostingsFormat {
@ -55,7 +54,7 @@ public class Elasticsearch090RWPostingsFormat extends Elasticsearch090PostingsFo
Fields maskedFields = new FilterLeafReader.FilterFields(fields) {
@Override
public Iterator<String> iterator() {
return Iterators.filter(this.in.iterator(), Predicates.not(UID_FIELD_FILTER));
return StreamSupport.stream(this.in.spliterator(), false).filter(UID_FIELD_FILTER.negate()).iterator();
}
};
fieldsConsumer.getDelegate().write(maskedFields);

16
core/src/test/java/org/elasticsearch/index/engine/InternalEngineMergeIT.java
View File

@ -18,7 +18,6 @@
*/
package org.elasticsearch.index.engine;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
@ -65,15 +64,12 @@ public class InternalEngineMergeIT extends ESIntegTestCase {
logger.info("index round [{}] - segments {}, total merges {}, current merge {}", i, stats.getPrimaries().getSegments().getCount(), stats.getPrimaries().getMerge().getTotal(), stats.getPrimaries().getMerge().getCurrent());
}
final long upperNumberSegments = 2 * numOfShards * 10;
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
IndicesStatsResponse stats = client().admin().indices().prepareStats().setSegments(true).setMerge(true).get();
logger.info("numshards {}, segments {}, total merges {}, current merge {}", numOfShards, stats.getPrimaries().getSegments().getCount(), stats.getPrimaries().getMerge().getTotal(), stats.getPrimaries().getMerge().getCurrent());
long current = stats.getPrimaries().getMerge().getCurrent();
long count = stats.getPrimaries().getSegments().getCount();
return count < upperNumberSegments && current == 0;
}
awaitBusy(() -> {
IndicesStatsResponse stats = client().admin().indices().prepareStats().setSegments(true).setMerge(true).get();
logger.info("numshards {}, segments {}, total merges {}, current merge {}", numOfShards, stats.getPrimaries().getSegments().getCount(), stats.getPrimaries().getMerge().getTotal(), stats.getPrimaries().getMerge().getCurrent());
long current = stats.getPrimaries().getMerge().getCurrent();
long count = stats.getPrimaries().getSegments().getCount();
return count < upperNumberSegments && current == 0;
});
IndicesStatsResponse stats = client().admin().indices().prepareStats().setSegments(true).setMerge(true).get();
logger.info("numshards {}, segments {}, total merges {}, current merge {}", numOfShards, stats.getPrimaries().getSegments().getCount(), stats.getPrimaries().getMerge().getTotal(), stats.getPrimaries().getMerge().getCurrent());

18
core/src/test/java/org/elasticsearch/index/store/CorruptedFileIT.java
View File

@ -20,7 +20,6 @@ package org.elasticsearch.index.store;
import com.carrotsearch.randomizedtesting.generators.RandomPicks;
import com.google.common.base.Charsets;
import com.google.common.base.Predicate;
import org.apache.lucene.codecs.CodecUtil;
import org.apache.lucene.index.CheckIndex;
import org.apache.lucene.index.IndexFileNames;
@ -107,8 +106,16 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.common.util.CollectionUtils.iterableAsArrayList;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*;
import static org.hamcrest.Matchers.*;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAllSuccessful;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.notNullValue;
@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.SUITE)
public class CorruptedFileIT extends ESIntegTestCase {
@ -279,13 +286,10 @@ public class CorruptedFileIT extends ESIntegTestCase {
client().admin().indices().prepareUpdateSettings("test").setSettings(build).get();
client().admin().cluster().prepareReroute().get();
boolean didClusterTurnRed = awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
boolean didClusterTurnRed = awaitBusy(() -> {
ClusterHealthStatus test = client().admin().cluster()
.health(Requests.clusterHealthRequest("test")).actionGet().getStatus();
return test == ClusterHealthStatus.RED;
}
}, 5, TimeUnit.MINUTES);// sometimes on slow nodes the replication / recovery is just dead slow
final ClusterHealthResponse response = client().admin().cluster()
.health(Requests.clusterHealthRequest("test")).get();

29
core/src/test/java/org/elasticsearch/indices/IndicesLifecycleListenerIT.java
View File

@ -18,7 +18,6 @@
*/
package org.elasticsearch.indices;
import com.google.common.base.Predicate;
import com.google.common.collect.Maps;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
@ -45,12 +44,17 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BooleanSupplier;
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS;
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
import static org.elasticsearch.cluster.routing.allocation.decider.DisableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_DISABLE_ALLOCATION;
import static org.elasticsearch.common.settings.Settings.builder;
import static org.elasticsearch.index.shard.IndexShardState.*;
import static org.elasticsearch.index.shard.IndexShardState.CLOSED;
import static org.elasticsearch.index.shard.IndexShardState.CREATED;
import static org.elasticsearch.index.shard.IndexShardState.POST_RECOVERY;
import static org.elasticsearch.index.shard.IndexShardState.RECOVERING;
import static org.elasticsearch.index.shard.IndexShardState.STARTED;
import static org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import static org.elasticsearch.test.ESIntegTestCase.Scope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
@ -203,24 +207,21 @@ public class IndicesLifecycleListenerIT extends ESIntegTestCase {
private static void assertShardStatesMatch(final IndexShardStateChangeListener stateChangeListener, final int numShards, final IndexShardState... shardStates)
throws InterruptedException {
Predicate<Object> waitPredicate = new Predicate<Object>() {
@Override
public boolean apply(Object input) {
if (stateChangeListener.shardStates.size() != numShards) {
BooleanSupplier waitPredicate = () -> {
if (stateChangeListener.shardStates.size() != numShards) {
return false;
}
for (List<IndexShardState> indexShardStates : stateChangeListener.shardStates.values()) {
if (indexShardStates == null || indexShardStates.size() != shardStates.length) {
return false;
}
for (List<IndexShardState> indexShardStates : stateChangeListener.shardStates.values()) {
if (indexShardStates == null || indexShardStates.size() != shardStates.length) {
for (int i = 0; i < shardStates.length; i++) {
if (indexShardStates.get(i) != shardStates[i]) {
return false;
}
for (int i = 0; i < shardStates.length; i++) {
if (indexShardStates.get(i) != shardStates[i]) {
return false;
}
}
}
return true;
}
return true;
};
if (!awaitBusy(waitPredicate, 1, TimeUnit.MINUTES)) {
fail("failed to observe expect shard states\n" +

39
core/src/test/java/org/elasticsearch/indices/memory/IndexingMemoryControllerIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.indices.memory;
import com.google.common.base.Predicate;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.settings.Settings;
@ -54,13 +53,9 @@ public class IndexingMemoryControllerIT extends ESIntegTestCase {
final long expected1ShardSize = internalCluster().getInstance(IndexingMemoryController.class).indexingBufferSize().bytes();
final long expected2ShardsSize = expected1ShardSize / 2;
boolean success = awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return shard1.engine().config().getIndexingBufferSize().bytes() <= expected2ShardsSize &&
shard2.engine().config().getIndexingBufferSize().bytes() <= expected2ShardsSize;
}
});
boolean success = awaitBusy(() -> shard1.engine().config().getIndexingBufferSize().bytes() <= expected2ShardsSize &&
shard2.engine().config().getIndexingBufferSize().bytes() <= expected2ShardsSize
);
if (!success) {
fail("failed to update shard indexing buffer size. expected [" + expected2ShardsSize + "] shard1 [" +
@ -70,12 +65,7 @@ public class IndexingMemoryControllerIT extends ESIntegTestCase {
}
client().admin().indices().prepareDelete("test2").get();
success = awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return shard1.engine().config().getIndexingBufferSize().bytes() >= expected1ShardSize;
}
});
success = awaitBusy(() -> shard1.engine().config().getIndexingBufferSize().bytes() >= expected1ShardSize);
if (!success) {
fail("failed to update shard indexing buffer size after deleting shards. expected [" + expected1ShardSize + "] got [" +
@ -95,12 +85,7 @@ public class IndexingMemoryControllerIT extends ESIntegTestCase {
ensureGreen();
final IndexShard shard1 = internalCluster().getInstance(IndicesService.class).indexService("test1").shard(0);
boolean success = awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return shard1.engine().config().getIndexingBufferSize().bytes() == EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER.bytes();
}
});
boolean success = awaitBusy(() -> shard1.engine().config().getIndexingBufferSize().bytes() == EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER.bytes());
if (!success) {
fail("failed to update shard indexing buffer size due to inactive state. expected [" + EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER + "] got [" +
shard1.engine().config().getIndexingBufferSize().bytes() + "]"
@ -109,12 +94,7 @@ public class IndexingMemoryControllerIT extends ESIntegTestCase {
index("test1", "type", "1", "f", 1);
success = awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return shard1.engine().config().getIndexingBufferSize().bytes() > EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER.bytes();
}
});
success = awaitBusy(() -> shard1.engine().config().getIndexingBufferSize().bytes() > EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER.bytes());
if (!success) {
fail("failed to update shard indexing buffer size due to inactive state. expected something larger then [" + EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER + "] got [" +
shard1.engine().config().getIndexingBufferSize().bytes() + "]"
@ -123,12 +103,7 @@ public class IndexingMemoryControllerIT extends ESIntegTestCase {
flush(); // clean translogs
success = awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return shard1.engine().config().getIndexingBufferSize().bytes() == EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER.bytes();
}
});
success = awaitBusy(() -> shard1.engine().config().getIndexingBufferSize().bytes() == EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER.bytes());
if (!success) {
fail("failed to update shard indexing buffer size due to inactive state. expected [" + EngineConfig.INACTIVE_SHARD_INDEXING_BUFFER + "] got [" +
shard1.engine().config().getIndexingBufferSize().bytes() + "]"

16
core/src/test/java/org/elasticsearch/indices/store/IndicesStoreIntegrationIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.indices.store;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
@ -55,7 +54,6 @@ import org.elasticsearch.test.disruption.BlockClusterStateProcessing;
import org.elasticsearch.test.disruption.SingleNodeDisruption;
import org.elasticsearch.test.junit.annotations.TestLogging;
import org.elasticsearch.test.transport.MockTransportService;
import org.elasticsearch.transport.TransportModule;
import org.elasticsearch.transport.TransportRequestOptions;
import org.elasticsearch.transport.TransportService;
import org.junit.Test;
@ -400,22 +398,12 @@ public class IndicesStoreIntegrationIT extends ESIntegTestCase {
}
private boolean waitForShardDeletion(final String server, final String index, final int shard) throws InterruptedException {
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
return !Files.exists(shardDirectory(server, index, shard));
}
});
awaitBusy(() -> !Files.exists(shardDirectory(server, index, shard)));
return Files.exists(shardDirectory(server, index, shard));
}
private boolean waitForIndexDeletion(final String server, final String index) throws InterruptedException {
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
return !Files.exists(indexDirectory(server, index));
}
});
awaitBusy(() -> !Files.exists(indexDirectory(server, index)));
return Files.exists(indexDirectory(server, index));
}

31
core/src/test/java/org/elasticsearch/percolator/RecoveryPercolatorIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.percolator;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.count.CountResponse;
@ -38,17 +37,27 @@ import org.junit.Test;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import static org.elasticsearch.action.percolate.PercolateSourceBuilder.docBuilder;
import static org.elasticsearch.client.Requests.clusterHealthRequest;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.index.query.QueryBuilders.*;
import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;
import static org.elasticsearch.percolator.PercolatorIT.convertFromTextArray;
import static org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import static org.elasticsearch.test.ESIntegTestCase.Scope;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*;
import static org.hamcrest.Matchers.*;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertMatchCount;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
import static org.hamcrest.Matchers.arrayWithSize;
import static org.hamcrest.Matchers.emptyArray;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.nullValue;
@ClusterScope(scope = Scope.TEST, numDataNodes = 0, numClientNodes = 0, transportClientRatio = 0)
public class RecoveryPercolatorIT extends ESIntegTestCase {
@ -239,12 +248,7 @@ public class RecoveryPercolatorIT extends ESIntegTestCase {
.get();
ensureGreen();
final Client client = internalCluster().client(new Predicate<Settings>() {
@Override
public boolean apply(Settings input) {
return input.getAsBoolean("node.stay", true);
}
});
final Client client = internalCluster().client(input -> input.getAsBoolean("node.stay", true));
final int numQueries = randomIntBetween(50, 100);
logger.info("--> register a queries");
for (int i = 0; i < numQueries; i++) {
@ -314,12 +318,7 @@ public class RecoveryPercolatorIT extends ESIntegTestCase {
};
Thread t = new Thread(r);
t.start();
Predicate<Settings> nodePredicate = new Predicate<Settings>() {
@Override
public boolean apply(Settings input) {
return !input.getAsBoolean("node.stay", false);
}
};
Predicate<Settings> nodePredicate = input -> !input.getAsBoolean("node.stay", false);
try {
// 1 index, 2 primaries, 2 replicas per primary
for (int i = 0; i < 4; i++) {

34
core/src/test/java/org/elasticsearch/percolator/TTLPercolatorIT.java
View File

@ -19,8 +19,6 @@
package org.elasticsearch.percolator;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.percolate.PercolateResponse;
@ -42,7 +40,9 @@ import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.percolator.PercolatorIT.convertFromTextArray;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertMatchCount;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.arrayContaining;
import static org.hamcrest.Matchers.emptyArray;
import static org.hamcrest.Matchers.equalTo;
/**
*/
@ -136,14 +136,11 @@ public class TTLPercolatorIT extends ESIntegTestCase {
// See comment in SimpleTTLTests
logger.info("Checking if the ttl purger has run");
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
IndicesStatsResponse indicesStatsResponse = client.admin().indices().prepareStats("test").clear().setIndexing(true).get();
// TTL deletes one doc, but it is indexed in the primary shard and replica shards
return indicesStatsResponse.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount() == test.dataCopies;
}
}, 5, TimeUnit.SECONDS), equalTo(true));
assertTrue(awaitBusy(() -> {
IndicesStatsResponse indicesStatsResponse = client.admin().indices().prepareStats("test").clear().setIndexing(true).get();
// TTL deletes one doc, but it is indexed in the primary shard and replica shards
return indicesStatsResponse.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount() == test.dataCopies;
}, 5, TimeUnit.SECONDS));
percolateResponse = client.preparePercolate()
.setIndices("test").setDocumentType("type1")
@ -200,15 +197,12 @@ public class TTLPercolatorIT extends ESIntegTestCase {
}
refresh();
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
IndicesStatsResponse indicesStatsResponse = client().admin().indices().prepareStats("test").clear().setIndexing(true).get();
logger.debug("delete count [{}]", indicesStatsResponse.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount());
// TTL deletes one doc, but it is indexed in the primary shard and replica shards
return indicesStatsResponse.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount() != 0;
}
}, 5, TimeUnit.SECONDS), equalTo(true));
assertTrue(awaitBusy(() -> {
IndicesStatsResponse indicesStatsResponse = client().admin().indices().prepareStats("test").clear().setIndexing(true).get();
logger.debug("delete count [{}]", indicesStatsResponse.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount());
// TTL deletes one doc, but it is indexed in the primary shard and replica shards
return indicesStatsResponse.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount() != 0;
}, 5, TimeUnit.SECONDS));
internalCluster().wipeIndices("test");
client().admin().indices().prepareCreate("test")
.addMapping("type1", typeMapping)

18
core/src/test/java/org/elasticsearch/recovery/RecoveriesCollectionTests.java
View File

@ -18,8 +18,6 @@
*/
package org.elasticsearch.recovery;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import org.elasticsearch.Version;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
@ -29,7 +27,11 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.indices.IndicesService;
import org.elasticsearch.indices.recovery.*;
import org.elasticsearch.indices.recovery.RecoveriesCollection;
import org.elasticsearch.indices.recovery.RecoveryFailedException;
import org.elasticsearch.indices.recovery.RecoveryState;
import org.elasticsearch.indices.recovery.RecoveryStatus;
import org.elasticsearch.indices.recovery.RecoveryTarget;
import org.elasticsearch.test.ESSingleNodeTestCase;
import org.elasticsearch.threadpool.ThreadPool;
import org.junit.Test;
@ -38,6 +40,7 @@ import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.lessThan;
@ -130,13 +133,8 @@ public class RecoveriesCollectionTests extends ESSingleNodeTestCase {
RecoveriesCollection.StatusRef statusRef = collection.getStatus(recoveryId);
toClose.add(statusRef);
ShardId shardId = statusRef.status().shardId();
assertFalse("should not have cancelled recoveries", collection.cancelRecoveriesForShard(shardId, "test", Predicates.<RecoveryStatus>alwaysFalse()));
final Predicate<RecoveryStatus> shouldCancel = new Predicate<RecoveryStatus>() {
@Override
public boolean apply(RecoveryStatus status) {
return status.recoveryId() == recoveryId;
}
};
assertFalse("should not have cancelled recoveries", collection.cancelRecoveriesForShard(shardId, "test", status -> false));
final Predicate<RecoveryStatus> shouldCancel = status -> status.recoveryId() == recoveryId;
assertTrue("failed to cancel recoveries", collection.cancelRecoveriesForShard(shardId, "test", shouldCancel));
assertThat("we should still have on recovery", collection.size(), equalTo(1));
statusRef = collection.getStatus(recoveryId);

29
core/src/test/java/org/elasticsearch/recovery/RecoveryWhileUnderLoadIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.recovery;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
import org.elasticsearch.action.admin.indices.stats.ShardStats;
@ -46,7 +45,6 @@ import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcke
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAllSuccessful;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoTimeout;
import static org.hamcrest.Matchers.equalTo;
public class RecoveryWhileUnderLoadIT extends ESIntegTestCase {
@ -289,19 +287,20 @@ public class RecoveryWhileUnderLoadIT extends ESIntegTestCase {
//if there was an error we try to wait and see if at some point it'll get fixed
logger.info("--> trying to wait");
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
boolean error = false;
for (int i = 0; i < iterations; i++) {
SearchResponse searchResponse = client().prepareSearch().setSize(0).setQuery(matchAllQuery()).get();
if (searchResponse.getHits().totalHits() != numberOfDocs) {
error = true;
}
}
return !error;
}
}, 5, TimeUnit.MINUTES), equalTo(true));
assertTrue(awaitBusy(() -> {
boolean errorOccurred = false;
for (int i = 0; i < iterations; i++) {
SearchResponse searchResponse = client().prepareSearch().setSize(0).setQuery(matchAllQuery()).get();
if (searchResponse.getHits().totalHits() != numberOfDocs) {
errorOccurred = true;
}
}
return !errorOccurred;
},
5,
TimeUnit.MINUTES
)
);
}
//lets now make the test fail if it was supposed to fail

12
core/src/test/java/org/elasticsearch/recovery/RelocationIT.java
View File

@ -21,7 +21,6 @@ package org.elasticsearch.recovery;
import com.carrotsearch.hppc.IntHashSet;
import com.carrotsearch.hppc.procedures.IntProcedure;
import com.google.common.base.Predicate;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.lucene.index.IndexFileNames;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
@ -416,13 +415,10 @@ public class RelocationIT extends ESIntegTestCase {
// Lets wait a bit and then move again to hopefully trigger recovery cancellations.
boolean applied = awaitBusy(
new Predicate<Object>() {
@Override
public boolean apply(Object input) {
RecoveryResponse recoveryResponse = internalCluster().client(redNodeName).admin().indices().prepareRecoveries(indexName)
.get();
return !recoveryResponse.shardRecoveryStates().get(indexName).isEmpty();
}
() -> {
RecoveryResponse recoveryResponse =
internalCluster().client(redNodeName).admin().indices().prepareRecoveries(indexName).get();
return !recoveryResponse.shardRecoveryStates().get(indexName).isEmpty();
}
);
assertTrue(applied);

10
core/src/test/java/org/elasticsearch/search/SearchWithRejectionsIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.search;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
@ -73,14 +72,7 @@ public class SearchWithRejectionsIT extends ESIntegTestCase {
} catch (Throwable t) {
}
}
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
// we must wait here because the requests to release search contexts might still be in flight
// although the search request has already returned
return client().admin().indices().prepareStats().execute().actionGet().getTotal().getSearch().getOpenContexts() == 0;
}
}, 1, TimeUnit.SECONDS);
awaitBusy(() -> client().admin().indices().prepareStats().execute().actionGet().getTotal().getSearch().getOpenContexts() == 0, 1, TimeUnit.SECONDS);
indicesStats = client().admin().indices().prepareStats().execute().actionGet();
assertThat(indicesStats.getTotal().getSearch().getOpenContexts(), equalTo(0l));
}

27
core/src/test/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java
View File

@ -18,7 +18,6 @@
*/
package org.elasticsearch.snapshots;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksResponse;
import org.elasticsearch.cluster.ClusterChangedEvent;
@ -49,6 +48,7 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.hamcrest.Matchers.equalTo;
@ -92,12 +92,7 @@ public abstract class AbstractSnapshotIntegTestCase extends ESIntegTestCase {
}
public static void stopNode(final String node) throws IOException {
internalCluster().stopRandomNode(new Predicate<Settings>() {
@Override
public boolean apply(Settings settings) {
return settings.get("name").equals(node);
}
});
internalCluster().stopRandomNode(settings -> settings.get("name").equals(node));
}
public void waitForBlock(String node, String repository, TimeValue timeout) throws InterruptedException {
@ -186,18 +181,8 @@ public abstract class AbstractSnapshotIntegTestCase extends ESIntegTestCase {
public BlockingClusterStateListener(ClusterService clusterService, final String blockOn, final String countOn, Priority passThroughPriority, TimeValue timeout) {
this.clusterService = clusterService;
this.blockOn = new Predicate<ClusterChangedEvent>() {
@Override
public boolean apply(ClusterChangedEvent clusterChangedEvent) {
return clusterChangedEvent.source().startsWith(blockOn);
}
};
this.countOn = new Predicate<ClusterChangedEvent>() {
@Override
public boolean apply(ClusterChangedEvent clusterChangedEvent) {
return clusterChangedEvent.source().startsWith(countOn);
}
};
this.blockOn = clusterChangedEvent -> clusterChangedEvent.source().startsWith(blockOn);
this.countOn = clusterChangedEvent -> clusterChangedEvent.source().startsWith(countOn);
this.latch = new CountDownLatch(1);
this.passThroughPriority = passThroughPriority;
this.timeout = timeout;
@ -210,13 +195,13 @@ public abstract class AbstractSnapshotIntegTestCase extends ESIntegTestCase {
@Override
public void clusterChanged(ClusterChangedEvent event) {
if (blockOn.apply(event)) {
if (blockOn.test(event)) {
logger.info("blocking cluster state tasks on [{}]", event.source());
assert stopWaitingAt < 0; // Make sure we are the first time here
stopWaitingAt = System.currentTimeMillis() + timeout.getMillis();
addBlock();
}
if (countOn.apply(event)) {
if (countOn.test(event)) {
count++;
}
}

35
core/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.snapshots;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableMap;
import org.apache.lucene.util.IOUtils;
import org.elasticsearch.ExceptionsHelper;
@ -84,8 +83,24 @@ import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
import static org.elasticsearch.index.shard.IndexShard.INDEX_REFRESH_INTERVAL;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*;
import static org.hamcrest.Matchers.*;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAliasesExist;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAliasesMissing;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAllSuccessful;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertBlocked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertIndexTemplateExists;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertIndexTemplateMissing;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertThrows;
import static org.hamcrest.Matchers.allOf;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.lessThan;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.startsWith;
public class SharedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTestCase {
@ -1857,21 +1872,11 @@ public class SharedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTestCas
}
private boolean waitForIndex(final String index, TimeValue timeout) throws InterruptedException {
return awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
return client().admin().indices().prepareExists(index).execute().actionGet().isExists();
}
}, timeout.millis(), TimeUnit.MILLISECONDS);
return awaitBusy(() -> client().admin().indices().prepareExists(index).execute().actionGet().isExists(), timeout.millis(), TimeUnit.MILLISECONDS);
}
private boolean waitForRelocationsToStart(final String index, TimeValue timeout) throws InterruptedException {
return awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object o) {
return client().admin().cluster().prepareHealth(index).execute().actionGet().getRelocatingShards() > 0;
}
}, timeout.millis(), TimeUnit.MILLISECONDS);
return awaitBusy(() -> client().admin().cluster().prepareHealth(index).execute().actionGet().getRelocatingShards() > 0, timeout.millis(), TimeUnit.MILLISECONDS);
}
@Test

14
core/src/test/java/org/elasticsearch/test/CompositeTestCluster.java
View File

@ -19,8 +19,6 @@
package org.elasticsearch.test;
import com.carrotsearch.randomizedtesting.generators.RandomPicks;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import com.google.common.collect.Iterators;
import org.apache.lucene.util.IOUtils;
import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
@ -33,10 +31,12 @@ import org.elasticsearch.common.transport.TransportAddress;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.Random;
import java.util.stream.Collectors;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoTimeout;
import static org.hamcrest.Matchers.equalTo;
@ -90,12 +90,10 @@ public class CompositeTestCluster extends TestCluster {
}
private Collection<ExternalNode> runningNodes() {
return Collections2.filter(Arrays.asList(externalNodes), new Predicate<ExternalNode>() {
@Override
public boolean apply(ExternalNode input) {
return input.running();
}
});
return Arrays
.stream(externalNodes)
.filter(input -> input.running())
.collect(Collectors.toCollection(ArrayList::new));
}
/**

63
core/src/test/java/org/elasticsearch/test/ESIntegTestCase.java
View File

@ -25,15 +25,7 @@ import com.carrotsearch.randomizedtesting.annotations.TestGroup;
import com.carrotsearch.randomizedtesting.generators.RandomInts;
import com.carrotsearch.randomizedtesting.generators.RandomPicks;
import com.google.common.base.Joiner;
import com.google.common.base.Predicate;
import org.apache.http.impl.client.HttpClients;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.UnassignedInfo;
import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
import org.elasticsearch.common.network.NetworkAddress;
import org.elasticsearch.index.shard.MergeSchedulerConfig;
import org.apache.lucene.util.IOUtils;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.lucene.util.TestUtil;
@ -63,6 +55,7 @@ import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.ClearScrollResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.Client;
@ -164,6 +157,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BooleanSupplier;
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS;
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
@ -172,8 +166,14 @@ import static org.elasticsearch.common.util.CollectionUtils.eagerPartition;
import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
import static org.elasticsearch.test.XContentTestUtils.convertToMap;
import static org.elasticsearch.test.XContentTestUtils.differenceBetweenMapsIgnoringArrayOrder;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*;
import static org.hamcrest.Matchers.*;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoTimeout;
import static org.hamcrest.Matchers.emptyIterable;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.startsWith;
/**
* {@link ESIntegTestCase} is an abstract base class to run integration
@ -1000,30 +1000,27 @@ public abstract class ESIntegTestCase extends ESTestCase {
throws InterruptedException {
final AtomicLong lastKnownCount = new AtomicLong(-1);
long lastStartCount = -1;
Predicate<Object> testDocs = new Predicate<Object>() {
@Override
public boolean apply(Object o) {
if (indexer != null) {
lastKnownCount.set(indexer.totalIndexedDocs());
}
if (lastKnownCount.get() >= numDocs) {
try {
long count = client().prepareCount().setQuery(matchAllQuery()).execute().actionGet().getCount();
if (count == lastKnownCount.get()) {
// no progress - try to refresh for the next time
client().admin().indices().prepareRefresh().get();
}
lastKnownCount.set(count);
} catch (Throwable e) { // count now acts like search and barfs if all shards failed...
logger.debug("failed to executed count", e);
return false;
}
logger.debug("[{}] docs visible for search. waiting for [{}]", lastKnownCount.get(), numDocs);
} else {
logger.debug("[{}] docs indexed. waiting for [{}]", lastKnownCount.get(), numDocs);
}
return lastKnownCount.get() >= numDocs;
BooleanSupplier testDocs = () -> {
if (indexer != null) {
lastKnownCount.set(indexer.totalIndexedDocs());
}
if (lastKnownCount.get() >= numDocs) {
try {
long count = client().prepareCount().setQuery(matchAllQuery()).execute().actionGet().getCount();
if (count == lastKnownCount.get()) {
// no progress - try to refresh for the next time
client().admin().indices().prepareRefresh().get();
}
lastKnownCount.set(count);
} catch (Throwable e) { // count now acts like search and barfs if all shards failed...
logger.debug("failed to executed count", e);
return false;
}
logger.debug("[{}] docs visible for search. waiting for [{}]", lastKnownCount.get(), numDocs);
} else {
logger.debug("[{}] docs indexed. waiting for [{}]", lastKnownCount.get(), numDocs);
}
return lastKnownCount.get() >= numDocs;
};
while (!awaitBusy(testDocs, maxWaitTime, maxWaitTimeUnit)) {

13
core/src/test/java/org/elasticsearch/test/ESTestCase.java
View File

@ -29,7 +29,6 @@ import com.carrotsearch.randomizedtesting.generators.RandomInts;
import com.carrotsearch.randomizedtesting.generators.RandomPicks;
import com.carrotsearch.randomizedtesting.generators.RandomStrings;
import com.carrotsearch.randomizedtesting.rules.TestRuleAdapter;
import com.google.common.base.Predicate;
import org.apache.lucene.uninverting.UninvertingReader;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.lucene.util.LuceneTestCase.SuppressCodecs;
@ -79,6 +78,8 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.BooleanSupplier;
import java.util.function.Predicate;
import static org.elasticsearch.common.util.CollectionUtils.arrayAsArrayList;
import static org.hamcrest.Matchers.equalTo;
@ -438,19 +439,19 @@ public abstract class ESTestCase extends LuceneTestCase {
}
}
public static boolean awaitBusy(Predicate<?> breakPredicate) throws InterruptedException {
return awaitBusy(breakPredicate, 10, TimeUnit.SECONDS);
public static boolean awaitBusy(BooleanSupplier breakSupplier) throws InterruptedException {
return awaitBusy(breakSupplier, 10, TimeUnit.SECONDS);
}
// After 1s, we stop growing the sleep interval exponentially and just sleep 1s until maxWaitTime
private static final long AWAIT_BUSY_THRESHOLD = 1000L;
public static boolean awaitBusy(Predicate<?> breakPredicate, long maxWaitTime, TimeUnit unit) throws InterruptedException {
public static boolean awaitBusy(BooleanSupplier breakSupplier, long maxWaitTime, TimeUnit unit) throws InterruptedException {
long maxTimeInMillis = TimeUnit.MILLISECONDS.convert(maxWaitTime, unit);
long timeInMillis = 1;
long sum = 0;
while (sum + timeInMillis < maxTimeInMillis) {
if (breakPredicate.apply(null)) {
if (breakSupplier.getAsBoolean()) {
return true;
}
Thread.sleep(timeInMillis);
@ -459,7 +460,7 @@ public abstract class ESTestCase extends LuceneTestCase {
}
timeInMillis = maxTimeInMillis - sum;
Thread.sleep(Math.max(timeInMillis, 0));
return breakPredicate.apply(null);
return breakSupplier.getAsBoolean();
}
public static boolean terminate(ExecutorService... services) throws InterruptedException {

19
core/src/test/java/org/elasticsearch/test/ExternalNode.java
View File

@ -18,8 +18,6 @@
*/
package org.elasticsearch.test;
import com.google.common.base.Predicate;
import org.apache.lucene.util.Constants;
import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
@ -155,18 +153,15 @@ final class ExternalNode implements Closeable {
}
static boolean waitForNode(final Client client, final String name) throws InterruptedException {
return ESTestCase.awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(java.lang.Object input) {
final NodesInfoResponse nodeInfos = client.admin().cluster().prepareNodesInfo().get();
final NodeInfo[] nodes = nodeInfos.getNodes();
for (NodeInfo info : nodes) {
if (name.equals(info.getNode().getName())) {
return true;
}
return ESTestCase.awaitBusy(() -> {
final NodesInfoResponse nodeInfos = client.admin().cluster().prepareNodesInfo().get();
final NodeInfo[] nodes = nodeInfos.getNodes();
for (NodeInfo info : nodes) {
if (name.equals(info.getNode().getName())) {
return true;
}
return false;
}
return false;
}, 30, TimeUnit.SECONDS);
}

96
core/src/test/java/org/elasticsearch/test/InternalTestCluster.java
View File

@ -24,12 +24,7 @@ import com.carrotsearch.randomizedtesting.SysGlobals;
import com.carrotsearch.randomizedtesting.generators.RandomInts;
import com.carrotsearch.randomizedtesting.generators.RandomPicks;
import com.carrotsearch.randomizedtesting.generators.RandomStrings;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.Collections2;
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
@ -127,13 +122,20 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static junit.framework.Assert.fail;
import static org.apache.lucene.util.LuceneTestCase.*;
import static org.apache.lucene.util.LuceneTestCase.TEST_NIGHTLY;
import static org.apache.lucene.util.LuceneTestCase.rarely;
import static org.apache.lucene.util.LuceneTestCase.usually;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.test.ESTestCase.assertBusy;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoTimeout;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.junit.Assert.assertThat;
/**
@ -532,14 +534,13 @@ public final class InternalTestCluster extends TestCluster {
}
private synchronized NodeAndClient getRandomNodeAndClient() {
Predicate<NodeAndClient> all = Predicates.alwaysTrue();
return getRandomNodeAndClient(all);
return getRandomNodeAndClient(nc -> true);
}
private synchronized NodeAndClient getRandomNodeAndClient(Predicate<NodeAndClient> predicate) {
ensureOpen();
Collection<NodeAndClient> values = Collections2.filter(nodes.values(), predicate);
Collection<NodeAndClient> values = nodes.values().stream().filter(predicate).collect(Collectors.toCollection(ArrayList::new));
if (!values.isEmpty()) {
int whichOne = random.nextInt(values.size());
for (NodeAndClient nodeAndClient : values) {
@ -588,8 +589,9 @@ public final class InternalTestCluster extends TestCluster {
return;
}
// prevent killing the master if possible and client nodes
final Iterator<NodeAndClient> values = n == 0 ? nodes.values().iterator() : Iterators.filter(nodes.values().iterator(),
Predicates.and(new DataNodePredicate(), Predicates.not(new MasterNodePredicate(getMasterName()))));
final Stream<NodeAndClient> collection =
n == 0 ? nodes.values().stream() : nodes.values().stream().filter(new DataNodePredicate().and(new MasterNodePredicate(getMasterName()).negate()));
final Iterator<NodeAndClient> values = collection.iterator();
final Iterator<NodeAndClient> limit = Iterators.limit(values, size - n);
logger.info("changing cluster size from {} to {}, {} data nodes", size(), n + numSharedClientNodes, n);
@ -682,7 +684,7 @@ public final class InternalTestCluster extends TestCluster {
*/
public synchronized Client nonMasterClient() {
ensureOpen();
NodeAndClient randomNodeAndClient = getRandomNodeAndClient(Predicates.not(new MasterNodePredicate(getMasterName())));
NodeAndClient randomNodeAndClient = getRandomNodeAndClient(new MasterNodePredicate(getMasterName()).negate());
if (randomNodeAndClient != null) {
return randomNodeAndClient.nodeClient(); // ensure node client non-master is requested
}
@ -758,12 +760,7 @@ public final class InternalTestCluster extends TestCluster {
*/
public synchronized Client client(final Predicate<Settings> filterPredicate) {
ensureOpen();
final NodeAndClient randomNodeAndClient = getRandomNodeAndClient(new Predicate<NodeAndClient>() {
@Override
public boolean apply(NodeAndClient nodeAndClient) {
return filterPredicate.apply(nodeAndClient.node.settings());
}
});
final NodeAndClient randomNodeAndClient = getRandomNodeAndClient(nodeAndClient -> filterPredicate.test(nodeAndClient.node.settings()));
if (randomNodeAndClient != null) {
return randomNodeAndClient.client(random);
}
@ -1145,7 +1142,7 @@ public final class InternalTestCluster extends TestCluster {
}
private synchronized <T> Iterable<T> getInstances(Class<T> clazz, Predicate<NodeAndClient> predicate) {
Iterable<NodeAndClient> filteredNodes = Iterables.filter(nodes.values(), predicate);
Iterable<NodeAndClient> filteredNodes = nodes.values().stream().filter(predicate)::iterator;
List<T> instances = new ArrayList<>();
for (NodeAndClient nodeAndClient : filteredNodes) {
instances.add(getInstanceFromNode(clazz, nodeAndClient.node));
@ -1157,18 +1154,7 @@ public final class InternalTestCluster extends TestCluster {
* Returns a reference to the given nodes instances of the given class &gt;T&lt;
*/
public synchronized <T> T getInstance(Class<T> clazz, final String node) {
final Predicate<InternalTestCluster.NodeAndClient> predicate;
if (node != null) {
predicate = new Predicate<InternalTestCluster.NodeAndClient>() {
@Override
public boolean apply(NodeAndClient nodeAndClient) {
return node.equals(nodeAndClient.name);
}
};
} else {
predicate = Predicates.alwaysTrue();
}
return getInstance(clazz, predicate);
return getInstance(clazz, nc -> node == null || node.equals(nc.name));
}
public synchronized <T> T getDataNodeInstance(Class<T> clazz) {
@ -1185,7 +1171,7 @@ public final class InternalTestCluster extends TestCluster {
* Returns a reference to a random nodes instances of the given class &gt;T&lt;
*/
public synchronized <T> T getInstance(Class<T> clazz) {
return getInstance(clazz, Predicates.<NodeAndClient>alwaysTrue());
return getInstance(clazz, nc -> true);
}
private synchronized <T> T getInstanceFromNode(Class<T> clazz, Node node) {
@ -1228,12 +1214,7 @@ public final class InternalTestCluster extends TestCluster {
*/
public synchronized void stopRandomNode(final Predicate<Settings> filter) throws IOException {
ensureOpen();
NodeAndClient nodeAndClient = getRandomNodeAndClient(new Predicate<InternalTestCluster.NodeAndClient>() {
@Override
public boolean apply(NodeAndClient nodeAndClient) {
return filter.apply(nodeAndClient.node.settings());
}
});
NodeAndClient nodeAndClient = getRandomNodeAndClient(nc -> filter.test(nc.node.settings()));
if (nodeAndClient != null) {
logger.info("Closing filtered random node [{}] ", nodeAndClient.name);
removeDisruptionSchemeFromNode(nodeAndClient);
@ -1260,7 +1241,7 @@ public final class InternalTestCluster extends TestCluster {
* Stops the any of the current nodes but not the master node.
*/
public void stopRandomNonMasterNode() throws IOException {
NodeAndClient nodeAndClient = getRandomNodeAndClient(Predicates.not(new MasterNodePredicate(getMasterName())));
NodeAndClient nodeAndClient = getRandomNodeAndClient(new MasterNodePredicate(getMasterName()).negate());
if (nodeAndClient != null) {
logger.info("Closing random non master node [{}] current master [{}] ", nodeAndClient.name, getMasterName());
removeDisruptionSchemeFromNode(nodeAndClient);
@ -1280,7 +1261,7 @@ public final class InternalTestCluster extends TestCluster {
* Restarts a random node in the cluster and calls the callback during restart.
*/
public void restartRandomNode(RestartCallback callback) throws Exception {
restartRandomNode(Predicates.<NodeAndClient>alwaysTrue(), callback);
restartRandomNode(nc -> true, callback);
}
/**
@ -1442,7 +1423,12 @@ public final class InternalTestCluster extends TestCluster {
private synchronized Set<String> nRandomDataNodes(int numNodes) {
assert size() >= numNodes;
NavigableMap<String, NodeAndClient> dataNodes = Maps.filterEntries(nodes, new EntryNodePredicate(new DataNodePredicate()));
Map<String, NodeAndClient> dataNodes =
nodes
.entrySet()
.stream()
.filter(new EntryNodePredicate(new DataNodePredicate()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
return Sets.newHashSet(Iterators.limit(dataNodes.keySet().iterator(), numNodes));
}
@ -1676,23 +1662,31 @@ public final class InternalTestCluster extends TestCluster {
}
private synchronized Collection<NodeAndClient> dataNodeAndClients() {
return Collections2.filter(nodes.values(), new DataNodePredicate());
return filterNodes(nodes, new DataNodePredicate());
}
private synchronized Collection<NodeAndClient> dataAndMasterNodes() {
return Collections2.filter(nodes.values(), new DataOrMasterNodePredicate());
return filterNodes(nodes, new DataOrMasterNodePredicate());
}
private synchronized Collection<NodeAndClient> filterNodes(Map<String, InternalTestCluster.NodeAndClient> map, Predicate<NodeAndClient> predicate) {
return map
.values()
.stream()
.filter(predicate)
.collect(Collectors.toCollection(ArrayList::new));
}
private static final class DataNodePredicate implements Predicate<NodeAndClient> {
@Override
public boolean apply(NodeAndClient nodeAndClient) {
public boolean test(NodeAndClient nodeAndClient) {
return DiscoveryNode.dataNode(nodeAndClient.node.settings());
}
}
private static final class DataOrMasterNodePredicate implements Predicate<NodeAndClient> {
@Override
public boolean apply(NodeAndClient nodeAndClient) {
public boolean test(NodeAndClient nodeAndClient) {
return DiscoveryNode.dataNode(nodeAndClient.node.settings()) ||
DiscoveryNode.masterNode(nodeAndClient.node.settings());
}
@ -1706,14 +1700,14 @@ public final class InternalTestCluster extends TestCluster {
}
@Override
public boolean apply(NodeAndClient nodeAndClient) {
public boolean test(NodeAndClient nodeAndClient) {
return masterNodeName.equals(nodeAndClient.name);
}
}
private static final class ClientNodePredicate implements Predicate<NodeAndClient> {
@Override
public boolean apply(NodeAndClient nodeAndClient) {
public boolean test(NodeAndClient nodeAndClient) {
return DiscoveryNode.clientNode(nodeAndClient.node.settings());
}
}
@ -1726,8 +1720,8 @@ public final class InternalTestCluster extends TestCluster {
}
@Override
public boolean apply(Map.Entry<String, NodeAndClient> entry) {
return delegateNodePredicate.apply(entry.getValue());
public boolean test(Map.Entry<String, NodeAndClient> entry) {
return delegateNodePredicate.test(entry.getValue());
}
}
@ -1795,7 +1789,7 @@ public final class InternalTestCluster extends TestCluster {
}
@Override
public boolean apply(Settings settings) {
public boolean test(Settings settings) {
return nodeNames.contains(settings.get("name"));
}

112
core/src/test/java/org/elasticsearch/test/hamcrest/ElasticsearchAssertions.java
View File

@ -18,11 +18,6 @@
*/
package org.elasticsearch.test.hamcrest;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.Iterables;
import org.apache.lucene.search.BooleanQuery;
import org.apache.lucene.search.Query;
import org.elasticsearch.ElasticsearchException;
@ -35,7 +30,6 @@ import org.elasticsearch.action.ShardOperationFailedException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequestBuilder;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.plugins.PluginInfo;
import org.elasticsearch.action.admin.cluster.node.info.PluginsInfo;
import org.elasticsearch.action.admin.indices.alias.exists.AliasesExistResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequestBuilder;
@ -62,6 +56,7 @@ import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.io.stream.Streamable;
import org.elasticsearch.plugins.PluginInfo;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.suggest.Suggest;
@ -77,16 +72,36 @@ import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Set;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import static com.google.common.base.Predicates.isNull;
import static com.google.common.base.Predicates.notNull;
import static org.elasticsearch.test.ESTestCase.*;
import static org.elasticsearch.test.ESTestCase.assertArrayEquals;
import static org.elasticsearch.test.ESTestCase.assertEquals;
import static org.elasticsearch.test.ESTestCase.assertFalse;
import static org.elasticsearch.test.ESTestCase.assertNotNull;
import static org.elasticsearch.test.ESTestCase.assertTrue;
import static org.elasticsearch.test.ESTestCase.fail;
import static org.elasticsearch.test.ESTestCase.random;
import static org.elasticsearch.test.VersionUtils.randomVersion;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.hasKey;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
/**
*
@ -739,91 +754,68 @@ public class ElasticsearchAssertions {
PluginsInfo plugins = response.getNodesMap().get(nodeId).getPlugins();
Assert.assertThat(plugins, notNullValue());
List<String> pluginNames = FluentIterable.from(plugins.getInfos()).filter(jvmPluginPredicate).transform(nameFunction).toList();
List<String> pluginNames = filterAndMap(plugins, jvmPluginPredicate, nameFunction);
for (String expectedJvmPluginName : expectedJvmPluginNames) {
Assert.assertThat(pluginNames, hasItem(expectedJvmPluginName));
}
List<String> pluginDescriptions = FluentIterable.from(plugins.getInfos()).filter(jvmPluginPredicate).transform(descriptionFunction).toList();
List<String> pluginDescriptions = filterAndMap(plugins, jvmPluginPredicate, descriptionFunction);
for (String expectedJvmPluginDescription : expectedJvmPluginDescriptions) {
Assert.assertThat(pluginDescriptions, hasItem(expectedJvmPluginDescription));
}
List<String> jvmPluginVersions = FluentIterable.from(plugins.getInfos()).filter(jvmPluginPredicate).transform(versionFunction).toList();
List<String> jvmPluginVersions = filterAndMap(plugins, jvmPluginPredicate, versionFunction);
for (String pluginVersion : expectedJvmVersions) {
Assert.assertThat(jvmPluginVersions, hasItem(pluginVersion));
}
FluentIterable<String> jvmUrls = FluentIterable.from(plugins.getInfos())
.filter(Predicates.and(jvmPluginPredicate, Predicates.not(sitePluginPredicate)))
.transform(urlFunction)
.filter(notNull());
Assert.assertThat(Iterables.size(jvmUrls), is(0));
boolean anyHaveUrls =
plugins
.getInfos()
.stream()
.filter(jvmPluginPredicate.and(sitePluginPredicate.negate()))
.map(urlFunction)
.anyMatch(p -> p != null);
assertFalse(anyHaveUrls);
List<String> sitePluginNames = filterAndMap(plugins, sitePluginPredicate, nameFunction);
List<String> sitePluginNames = FluentIterable.from(plugins.getInfos()).filter(sitePluginPredicate).transform(nameFunction).toList();
Assert.assertThat(sitePluginNames.isEmpty(), is(expectedSitePluginNames.isEmpty()));
for (String expectedSitePluginName : expectedSitePluginNames) {
Assert.assertThat(sitePluginNames, hasItem(expectedSitePluginName));
}
List<String> sitePluginDescriptions = FluentIterable.from(plugins.getInfos()).filter(sitePluginPredicate).transform(descriptionFunction).toList();
List<String> sitePluginDescriptions = filterAndMap(plugins, sitePluginPredicate, descriptionFunction);
Assert.assertThat(sitePluginDescriptions.isEmpty(), is(expectedSitePluginDescriptions.isEmpty()));
for (String sitePluginDescription : expectedSitePluginDescriptions) {
Assert.assertThat(sitePluginDescriptions, hasItem(sitePluginDescription));
}
List<String> sitePluginUrls = FluentIterable.from(plugins.getInfos()).filter(sitePluginPredicate).transform(urlFunction).toList();
List<String> sitePluginUrls = filterAndMap(plugins, sitePluginPredicate, urlFunction);
Assert.assertThat(sitePluginUrls, not(contains(nullValue())));
List<String> sitePluginVersions = FluentIterable.from(plugins.getInfos()).filter(sitePluginPredicate).transform(versionFunction).toList();
List<String> sitePluginVersions = filterAndMap(plugins, sitePluginPredicate, versionFunction);
Assert.assertThat(sitePluginVersions.isEmpty(), is(expectedSiteVersions.isEmpty()));
for (String pluginVersion : expectedSiteVersions) {
Assert.assertThat(sitePluginVersions, hasItem(pluginVersion));
}
}
private static Predicate<PluginInfo> jvmPluginPredicate = new Predicate<PluginInfo>() {
@Override
public boolean apply(PluginInfo pluginInfo) {
return pluginInfo.isJvm();
}
};
private static List<String> filterAndMap(PluginsInfo pluginsInfo, Predicate<PluginInfo> predicate, Function<PluginInfo, String> function) {
return pluginsInfo.getInfos().stream().filter(predicate).map(function).collect(Collectors.toList());
}
private static Predicate<PluginInfo> sitePluginPredicate = new Predicate<PluginInfo>() {
@Override
public boolean apply(PluginInfo pluginInfo) {
return pluginInfo.isSite();
}
};
private static Predicate<PluginInfo> jvmPluginPredicate = p -> p.isJvm();
private static Function<PluginInfo, String> nameFunction = new Function<PluginInfo, String>() {
@Override
public String apply(PluginInfo pluginInfo) {
return pluginInfo.getName();
}
};
private static Predicate<PluginInfo> sitePluginPredicate = p -> p.isSite();
private static Function<PluginInfo, String> descriptionFunction = new Function<PluginInfo, String>() {
@Override
public String apply(PluginInfo pluginInfo) {
return pluginInfo.getDescription();
}
};
private static Function<PluginInfo, String> nameFunction = p -> p.getName();
private static Function<PluginInfo, String> urlFunction = new Function<PluginInfo, String>() {
@Override
public String apply(PluginInfo pluginInfo) {
return pluginInfo.getUrl();
}
};
private static Function<PluginInfo, String> descriptionFunction = p -> p.getDescription();
private static Function<PluginInfo, String> versionFunction = new Function<PluginInfo, String>() {
@Override
public String apply(PluginInfo pluginInfo) {
return pluginInfo.getVersion();
}
};
private static Function<PluginInfo, String> urlFunction = p -> p.getUrl();
private static Function<PluginInfo, String> versionFunction = p -> p.getVersion();
/**
* Check if a file exists

29
core/src/test/java/org/elasticsearch/ttl/SimpleTTLIT.java
View File

@ -19,8 +19,6 @@
package org.elasticsearch.ttl;
import com.google.common.base.Predicate;
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
@ -31,7 +29,6 @@ import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.get.GetField;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
import org.elasticsearch.test.ESIntegTestCase.Scope;
@ -44,7 +41,6 @@ import java.util.concurrent.TimeUnit;
import static org.elasticsearch.common.settings.Settings.settingsBuilder;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
import static org.hamcrest.Matchers.both;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
@ -170,19 +166,18 @@ public class SimpleTTLIT extends ESIntegTestCase {
// But we can use index statistics' delete count to be sure that deletes have been executed, that must be incremented before
// ttl purging has finished.
logger.info("--> checking purger");
assertThat(awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
if (rarely()) {
client().admin().indices().prepareFlush("test").get();
} else if (rarely()) {
client().admin().indices().prepareOptimize("test").setMaxNumSegments(1).get();
}
IndicesStatsResponse response = client().admin().indices().prepareStats("test").clear().setIndexing(true).get();
// TTL deletes two docs, but it is indexed in the primary shard and replica shard.
return response.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount() == 2L * test.dataCopies;
}
}, 5, TimeUnit.SECONDS), equalTo(true));
assertTrue(awaitBusy(() -> {
if (rarely()) {
client().admin().indices().prepareFlush("test").get();
} else if (rarely()) {
client().admin().indices().prepareOptimize("test").setMaxNumSegments(1).get();
}
IndicesStatsResponse indicesStatsResponse = client().admin().indices().prepareStats("test").clear().setIndexing(true).get();
// TTL deletes two docs, but it is indexed in the primary shard and replica shard.
return indicesStatsResponse.getIndices().get("test").getTotal().getIndexing().getTotal().getDeleteCount() == 2L * test.dataCopies;
},
5, TimeUnit.SECONDS
));
// realtime get check
getResponse = client().prepareGet("test", "type1", "1").setFields("_ttl").setRealtime(true).execute().actionGet();

2
dev-tools/src/main/resources/forbidden/core-signatures.txt
View File

@ -88,3 +88,5 @@ org.elasticsearch.common.io.PathUtils#get(java.net.URI)
com.google.common.collect.Lists
com.google.common.collect.ImmutableList
com.google.common.base.Objects
com.google.common.base.Predicate
com.google.common.base.Predicates

17
plugins/delete-by-query/src/test/java/org/elasticsearch/action/deletebyquery/TransportDeleteByQueryActionTests.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.action.deletebyquery;
import com.google.common.base.Predicate;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ShardOperationFailedException;
import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
@ -164,13 +163,8 @@ public class TransportDeleteByQueryActionTests extends ESSingleNodeTestCase {
TestActionListener listener = new TestActionListener();
final TransportDeleteByQueryAction.AsyncDeleteByQueryAction async = newAsyncAction(delete, listener);
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
// Wait until the action timed out
return async.hasTimedOut();
}
});
// Wait until the action timed out
awaitBusy(() -> async.hasTimedOut());
async.executeScroll(searchResponse.getScrollId());
waitForCompletion("scroll request returns zero documents on expired scroll id", listener);
@ -419,12 +413,7 @@ public class TransportDeleteByQueryActionTests extends ESSingleNodeTestCase {
private void waitForCompletion(String testName, final TestActionListener listener) {
logger.info(" --> waiting for delete-by-query [{}] to complete", testName);
try {
awaitBusy(new Predicate<Object>() {
@Override
public boolean apply(Object input) {
return listener.isTerminated();
}
});
awaitBusy(() -> listener.isTerminated());
} catch (InterruptedException e) {
fail("exception when waiting for delete-by-query [" + testName + "] to complete: " + e.getMessage());
logger.error("exception when waiting for delete-by-query [{}] to complete", e, testName);