entry = entryItr.next();
+ if (entry.getValue() == null) {
+ // a null value obviously can't be replaced
+ continue;
+ }
+ String value = propertyPlaceholder.replacePlaceholders(entry.getValue(), placeholderResolver);
// if the values exists and has length, we should maintain it in the map
// otherwise, the replace process resolved into removing it
if (Strings.hasLength(value)) {
- map.put(entry.getKey(), value);
+ entry.setValue(value);
} else {
- map.remove(entry.getKey());
+ entryItr.remove();
}
}
return this;
diff --git a/core/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java b/core/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java
index 2d45a6fecff..5ac94f8b386 100644
--- a/core/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java
+++ b/core/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java
@@ -83,13 +83,16 @@ public class EsExecutors {
}
public static String threadName(Settings settings, String namePrefix) {
- String name = settings.get("node.name");
- if (name == null) {
- name = "elasticsearch";
+ String nodeName = settings.get("node.name");
+ if (nodeName == null) {
+ return threadName("", namePrefix);
} else {
- name = "elasticsearch[" + name + "]";
+ return threadName(nodeName, namePrefix);
}
- return name + "[" + namePrefix + "]";
+ }
+
+ public static String threadName(final String nodeName, final String namePrefix) {
+ return "elasticsearch" + (nodeName.isEmpty() ? "" : "[") + nodeName + (nodeName.isEmpty() ? "" : "]") + "[" + namePrefix + "]";
}
public static ThreadFactory daemonThreadFactory(Settings settings, String namePrefix) {
diff --git a/core/src/main/java/org/elasticsearch/common/xcontent/ConstructingObjectParser.java b/core/src/main/java/org/elasticsearch/common/xcontent/ConstructingObjectParser.java
index 3c4d8875d2a..e1400463a72 100644
--- a/core/src/main/java/org/elasticsearch/common/xcontent/ConstructingObjectParser.java
+++ b/core/src/main/java/org/elasticsearch/common/xcontent/ConstructingObjectParser.java
@@ -26,7 +26,6 @@ import org.elasticsearch.common.xcontent.ObjectParser.ValueType;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -53,12 +52,13 @@ import java.util.function.Function;
*
* {@code
* private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>("thing",
- * a -> new Thing((String) a[0], (String) a[1]));
+ * a -> new Thing((String) a[0], (String) a[1], (Integer) a[2]));
* static {
* PARSER.declareString(constructorArg(), new ParseField("animal"));
* PARSER.declareString(constructorArg(), new ParseField("vegetable"));
- * PARSER.declareInt(Thing::setMineral, new ParseField("mineral"));
+ * PARSER.declareInt(optionalConstructorArg(), new ParseField("mineral"));
* PARSER.declareInt(Thing::setFruit, new ParseField("fruit"));
+ * PARSER.declareInt(Thing::setBug, new ParseField("bug"));
* }
* }
*
@@ -70,19 +70,29 @@ import java.util.function.Function;
* it allocates 3 + 2 * param_count objects per parse. If this overhead is too much for you then feel free to have ObjectParser
* parse a secondary object and have that one call the target object's constructor. That ought to be rare though.
*
+ *
+ * Note: if optional constructor arguments aren't specified then the number of allocations is always the worst case.
+ *
*/
public final class ConstructingObjectParser extends AbstractObjectParser {
/**
- * Consumer that marks a field as a constructor argument instead of a real object field.
+ * Consumer that marks a field as a required constructor argument instead of a real object field.
*/
- private static final BiConsumer CONSTRUCTOR_ARG_MARKER = (a, b) -> {
+ private static final BiConsumer REQUIRED_CONSTRUCTOR_ARG_MARKER = (a, b) -> {
+ throw new UnsupportedOperationException("I am just a marker I should never be called.");
+ };
+
+ /**
+ * Consumer that marks a field as an optional constructor argument instead of a real object field.
+ */
+ private static final BiConsumer OPTIONAL_CONSTRUCTOR_ARG_MARKER = (a, b) -> {
throw new UnsupportedOperationException("I am just a marker I should never be called.");
};
/**
* List of constructor names used for generating the error message if not all arrive.
*/
- private final List constructorArgNames = new ArrayList<>();
+ private final List constructorArgInfos = new ArrayList<>();
private final ObjectParser objectParser;
private final Function builder;
/**
@@ -120,27 +130,39 @@ public final class ConstructingObjectParser BiConsumer constructorArg() {
- return (BiConsumer) CONSTRUCTOR_ARG_MARKER;
+ return (BiConsumer) REQUIRED_CONSTRUCTOR_ARG_MARKER;
+ }
+
+ /**
+ * Pass the {@linkplain BiConsumer} this returns the declare methods to declare an optional constructor argument. See this class's
+ * javadoc for an example. The order in which these are declared matters: it is the order that they come in the array passed to
+ * {@link #builder} and the order that missing arguments are reported to the user if any are missing. When all of these parameters are
+ * parsed from the {@linkplain XContentParser} the target object is immediately built.
+ */
+ @SuppressWarnings("unchecked") // Safe because we never call the method. This is just trickery to make the interface pretty.
+ public static BiConsumer optionalConstructorArg() {
+ return (BiConsumer) OPTIONAL_CONSTRUCTOR_ARG_MARKER;
}
@Override
public void declareField(BiConsumer consumer, ContextParser parser, ParseField parseField, ValueType type) {
- if (consumer == CONSTRUCTOR_ARG_MARKER) {
+ if (consumer == REQUIRED_CONSTRUCTOR_ARG_MARKER || consumer == OPTIONAL_CONSTRUCTOR_ARG_MARKER) {
/*
* Constructor arguments are detected by this "marker" consumer. It keeps the API looking clean even if it is a bit sleezy. We
* then build a new consumer directly against the object parser that triggers the "constructor arg just arrived behavior" of the
* parser. Conveniently, we can close over the position of the constructor in the argument list so we don't need to do any fancy
* or expensive lookups whenever the constructor args come in.
*/
- int position = constructorArgNames.size();
- constructorArgNames.add(parseField);
+ int position = constructorArgInfos.size();
+ boolean required = consumer == REQUIRED_CONSTRUCTOR_ARG_MARKER;
+ constructorArgInfos.add(new ConstructorArgInfo(parseField, required));
objectParser.declareField((target, v) -> target.constructorArg(position, parseField, v), parser, parseField, type);
} else {
numberOfFields += 1;
@@ -186,7 +208,7 @@ public final class ConstructingObjectParser 0) {
- queuedFieldsCount -= 1;
- queuedFields[queuedFieldsCount].accept(targetObject);
- }
- } catch (ParsingException e) {
- throw new ParsingException(e.getLineNumber(), e.getColumnNumber(),
- "failed to build [" + objectParser.getName() + "] after last required field arrived", e);
- } catch (Exception e) {
- throw new ParsingException(null, "Failed to build [" + objectParser.getName() + "] after last required field arrived", e);
+ if (constructorArgsCollected == constructorArgInfos.size()) {
+ buildTarget();
}
}
@@ -263,36 +273,62 @@ public final class ConstructingObjectParser 0) {
+ queuedFieldsCount -= 1;
+ queuedFields[queuedFieldsCount].accept(targetObject);
+ }
+ } catch (ParsingException e) {
+ throw new ParsingException(e.getLineNumber(), e.getColumnNumber(),
+ "failed to build [" + objectParser.getName() + "] after last required field arrived", e);
+ } catch (Exception e) {
+ throw new ParsingException(null, "Failed to build [" + objectParser.getName() + "] after last required field arrived", e);
}
- if (message == null) {
- throw new IllegalStateException("The targetObject wasn't built but we aren't missing any constructor args. This is a bug "
- + " in ConstructingObjectParser. Here are the constructor arguments " + Arrays.toString(constructorArgs)
- + " and here are is the count [" + constructorArgsCollected + "]. Good luck figuring out what happened."
- + " I'm truly sorry you got here.");
- }
- throw new IllegalArgumentException(message.append(']').toString());
+ }
+ }
+
+ private static class ConstructorArgInfo {
+ final ParseField field;
+ final boolean required;
+
+ public ConstructorArgInfo(ParseField field, boolean required) {
+ this.field = field;
+ this.required = required;
}
}
}
diff --git a/core/src/main/java/org/elasticsearch/env/Environment.java b/core/src/main/java/org/elasticsearch/env/Environment.java
index c44db111a5a..6bec05fc82b 100644
--- a/core/src/main/java/org/elasticsearch/env/Environment.java
+++ b/core/src/main/java/org/elasticsearch/env/Environment.java
@@ -195,7 +195,11 @@ public class Environment {
/**
* The data location with the cluster name as a sub directory.
+ *
+ * @deprecated Used to upgrade old data paths to new ones that do not include the cluster name, should not be used to write files to and
+ * will be removed in ES 6.0
*/
+ @Deprecated
public Path[] dataWithClusterFiles() {
return dataWithClusterFiles;
}
diff --git a/core/src/main/java/org/elasticsearch/env/NodeEnvironment.java b/core/src/main/java/org/elasticsearch/env/NodeEnvironment.java
index 98b47d92246..2b2aab2ccd2 100644
--- a/core/src/main/java/org/elasticsearch/env/NodeEnvironment.java
+++ b/core/src/main/java/org/elasticsearch/env/NodeEnvironment.java
@@ -32,9 +32,12 @@ import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.SuppressForbidden;
+import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.FileSystemUtils;
+import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.settings.Setting;
import org.elasticsearch.common.settings.Setting.Property;
import org.elasticsearch.common.settings.Settings;
@@ -63,6 +66,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -161,6 +165,7 @@ public final class NodeEnvironment extends AbstractComponent implements Closeabl
public static final String NODES_FOLDER = "nodes";
public static final String INDICES_FOLDER = "indices";
public static final String NODE_LOCK_FILENAME = "node.lock";
+ public static final String UPGRADE_LOCK_FILENAME = "upgrade.lock";
@Inject
public NodeEnvironment(Settings settings, Environment environment) throws IOException {
@@ -175,7 +180,7 @@ public final class NodeEnvironment extends AbstractComponent implements Closeabl
localNodeId = -1;
return;
}
- final NodePath[] nodePaths = new NodePath[environment.dataWithClusterFiles().length];
+ final NodePath[] nodePaths = new NodePath[environment.dataFiles().length];
final Lock[] locks = new Lock[nodePaths.length];
boolean success = false;
@@ -185,8 +190,17 @@ public final class NodeEnvironment extends AbstractComponent implements Closeabl
IOException lastException = null;
int maxLocalStorageNodes = MAX_LOCAL_STORAGE_NODES_SETTING.get(settings);
for (int possibleLockId = 0; possibleLockId < maxLocalStorageNodes; possibleLockId++) {
- for (int dirIndex = 0; dirIndex < environment.dataWithClusterFiles().length; dirIndex++) {
- Path dir = environment.dataWithClusterFiles()[dirIndex].resolve(NODES_FOLDER).resolve(Integer.toString(possibleLockId));
+ for (int dirIndex = 0; dirIndex < environment.dataFiles().length; dirIndex++) {
+ Path dataDirWithClusterName = environment.dataWithClusterFiles()[dirIndex];
+ Path dataDir = environment.dataFiles()[dirIndex];
+ // TODO: Remove this in 6.0, we are no longer going to read from the cluster name directory
+ if (readFromDataPathWithClusterName(dataDirWithClusterName)) {
+ DeprecationLogger deprecationLogger = new DeprecationLogger(logger);
+ deprecationLogger.deprecated("ES has detected the [path.data] folder using the cluster name as a folder [{}], " +
+ "Elasticsearch 6.0 will not allow the cluster name as a folder within the data path", dataDir);
+ dataDir = dataDirWithClusterName;
+ }
+ Path dir = dataDir.resolve(NODES_FOLDER).resolve(Integer.toString(possibleLockId));
Files.createDirectories(dir);
try (Directory luceneDir = FSDirectory.open(dir, NativeFSLockFactory.INSTANCE)) {
@@ -218,7 +232,7 @@ public final class NodeEnvironment extends AbstractComponent implements Closeabl
if (locks[0] == null) {
throw new IllegalStateException("Failed to obtain node lock, is the following location writable?: "
- + Arrays.toString(environment.dataWithClusterFiles()), lastException);
+ + Arrays.toString(environment.dataFiles()), lastException);
}
this.localNodeId = localNodeId;
@@ -242,6 +256,32 @@ public final class NodeEnvironment extends AbstractComponent implements Closeabl
}
}
+ /** Returns true if the directory is empty */
+ private static boolean dirEmpty(final Path path) throws IOException {
+ try (DirectoryStream stream = Files.newDirectoryStream(path)) {
+ return stream.iterator().hasNext() == false;
+ }
+ }
+
+ // Visible for testing
+ /** Returns true if data should be read from the data path that includes the cluster name (ie, it has data in it) */
+ static boolean readFromDataPathWithClusterName(Path dataPathWithClusterName) throws IOException {
+ if (Files.exists(dataPathWithClusterName) == false || // If it doesn't exist
+ Files.isDirectory(dataPathWithClusterName) == false || // Or isn't a directory
+ dirEmpty(dataPathWithClusterName)) { // Or if it's empty
+ // No need to read from cluster-name folder!
+ return false;
+ }
+ // The "nodes" directory inside of the cluster name
+ Path nodesPath = dataPathWithClusterName.resolve(NODES_FOLDER);
+ if (Files.isDirectory(nodesPath)) {
+ // The cluster has data in the "nodes" so we should read from the cluster-named folder for now
+ return true;
+ }
+ // Hey the nodes directory didn't exist, so we can safely use whatever directory we feel appropriate
+ return false;
+ }
+
private static void releaseAndNullLocks(Lock[] locks) {
for (int i = 0; i < locks.length; i++) {
if (locks[i] != null) {
diff --git a/core/src/main/java/org/elasticsearch/index/IndexSettings.java b/core/src/main/java/org/elasticsearch/index/IndexSettings.java
index 7c8cb4ff8c8..592c1ff1125 100644
--- a/core/src/main/java/org/elasticsearch/index/IndexSettings.java
+++ b/core/src/main/java/org/elasticsearch/index/IndexSettings.java
@@ -115,6 +115,11 @@ public final class IndexSettings {
public static final Setting INDEX_GC_DELETES_SETTING =
Setting.timeSetting("index.gc_deletes", DEFAULT_GC_DELETES, new TimeValue(-1, TimeUnit.MILLISECONDS), Property.Dynamic,
Property.IndexScope);
+ /**
+ * The maximum number of refresh listeners allows on this shard.
+ */
+ public static final Setting MAX_REFRESH_LISTENERS_PER_SHARD = Setting.intSetting("index.max_refresh_listeners", 1000, 0,
+ Property.Dynamic, Property.IndexScope);
private final Index index;
private final Version version;
@@ -145,6 +150,10 @@ public final class IndexSettings {
private volatile int maxResultWindow;
private volatile int maxRescoreWindow;
private volatile boolean TTLPurgeDisabled;
+ /**
+ * The maximum number of refresh listeners allows on this shard.
+ */
+ private volatile int maxRefreshListeners;
/**
* Returns the default search field for this index.
@@ -229,6 +238,7 @@ public final class IndexSettings {
maxResultWindow = scopedSettings.get(MAX_RESULT_WINDOW_SETTING);
maxRescoreWindow = scopedSettings.get(MAX_RESCORE_WINDOW_SETTING);
TTLPurgeDisabled = scopedSettings.get(INDEX_TTL_DISABLE_PURGE_SETTING);
+ maxRefreshListeners = scopedSettings.get(MAX_REFRESH_LISTENERS_PER_SHARD);
this.mergePolicyConfig = new MergePolicyConfig(logger, this);
assert indexNameMatcher.test(indexMetaData.getIndex().getName());
@@ -251,6 +261,7 @@ public final class IndexSettings {
scopedSettings.addSettingsUpdateConsumer(INDEX_GC_DELETES_SETTING, this::setGCDeletes);
scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING, this::setTranslogFlushThresholdSize);
scopedSettings.addSettingsUpdateConsumer(INDEX_REFRESH_INTERVAL_SETTING, this::setRefreshInterval);
+ scopedSettings.addSettingsUpdateConsumer(MAX_REFRESH_LISTENERS_PER_SHARD, this::setMaxRefreshListeners);
}
private void setTranslogFlushThresholdSize(ByteSizeValue byteSizeValue) {
@@ -499,6 +510,16 @@ public final class IndexSettings {
return scopedSettings.get(setting);
}
+ /**
+ * The maximum number of refresh listeners allows on this shard.
+ */
+ public int getMaxRefreshListeners() {
+ return maxRefreshListeners;
+ }
+
+ private void setMaxRefreshListeners(int maxRefreshListeners) {
+ this.maxRefreshListeners = maxRefreshListeners;
+ }
IndexScopedSettings getScopedSettings() { return scopedSettings;}
}
diff --git a/core/src/main/java/org/elasticsearch/index/engine/Engine.java b/core/src/main/java/org/elasticsearch/index/engine/Engine.java
index ab142f9dd51..87ffb9331a6 100644
--- a/core/src/main/java/org/elasticsearch/index/engine/Engine.java
+++ b/core/src/main/java/org/elasticsearch/index/engine/Engine.java
@@ -607,6 +607,7 @@ public abstract class Engine implements Closeable {
* Synchronously refreshes the engine for new search operations to reflect the latest
* changes.
*/
+ @Nullable
public abstract void refresh(String source) throws EngineException;
/**
@@ -999,6 +1000,9 @@ public abstract class Engine implements Closeable {
public static final GetResult NOT_EXISTS = new GetResult(false, Versions.NOT_FOUND, null);
+ /**
+ * Build a realtime get result from the translog.
+ */
public GetResult(boolean exists, long version, @Nullable Translog.Source source) {
this.source = source;
this.exists = exists;
@@ -1007,6 +1011,9 @@ public abstract class Engine implements Closeable {
this.searcher = null;
}
+ /**
+ * Build a non-realtime get result from the searcher.
+ */
public GetResult(Searcher searcher, Versions.DocIdAndVersion docIdAndVersion) {
this.exists = true;
this.source = null;
diff --git a/core/src/main/java/org/elasticsearch/index/engine/EngineConfig.java b/core/src/main/java/org/elasticsearch/index/engine/EngineConfig.java
index 8a56feff70f..13408408e7e 100644
--- a/core/src/main/java/org/elasticsearch/index/engine/EngineConfig.java
+++ b/core/src/main/java/org/elasticsearch/index/engine/EngineConfig.java
@@ -25,14 +25,15 @@ import org.apache.lucene.index.SnapshotDeletionPolicy;
import org.apache.lucene.search.QueryCache;
import org.apache.lucene.search.QueryCachingPolicy;
import org.apache.lucene.search.similarities.Similarity;
+import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.settings.Setting;
import org.elasticsearch.common.settings.Setting.Property;
-import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.IndexSettings;
import org.elasticsearch.index.codec.CodecService;
+import org.elasticsearch.index.shard.RefreshListeners;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.shard.TranslogRecoveryPerformer;
import org.elasticsearch.index.store.Store;
@@ -40,8 +41,6 @@ import org.elasticsearch.index.translog.TranslogConfig;
import org.elasticsearch.indices.IndexingMemoryController;
import org.elasticsearch.threadpool.ThreadPool;
-import java.util.function.Function;
-
/*
* Holds all the configuration that is used to create an {@link Engine}.
* Once {@link Engine} has been created with this object, changes to this
@@ -66,6 +65,8 @@ public final class EngineConfig {
private final Engine.EventListener eventListener;
private final QueryCache queryCache;
private final QueryCachingPolicy queryCachingPolicy;
+ @Nullable
+ private final RefreshListeners refreshListeners;
/**
* Index setting to change the low level lucene codec used for writing new segments.
@@ -99,7 +100,7 @@ public final class EngineConfig {
MergePolicy mergePolicy,Analyzer analyzer,
Similarity similarity, CodecService codecService, Engine.EventListener eventListener,
TranslogRecoveryPerformer translogRecoveryPerformer, QueryCache queryCache, QueryCachingPolicy queryCachingPolicy,
- TranslogConfig translogConfig, TimeValue flushMergesAfter) {
+ TranslogConfig translogConfig, TimeValue flushMergesAfter, RefreshListeners refreshListeners) {
if (openMode == null) {
throw new IllegalArgumentException("openMode must not be null");
}
@@ -125,6 +126,7 @@ public final class EngineConfig {
this.translogConfig = translogConfig;
this.flushMergesAfter = flushMergesAfter;
this.openMode = openMode;
+ this.refreshListeners = refreshListeners;
}
/**
@@ -303,4 +305,10 @@ public final class EngineConfig {
OPEN_INDEX_AND_TRANSLOG;
}
+ /**
+ * {@linkplain RefreshListeners} instance to configure.
+ */
+ public RefreshListeners getRefreshListeners() {
+ return refreshListeners;
+ }
}
diff --git a/core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java b/core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
index c120b07fce8..15667e79421 100644
--- a/core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
+++ b/core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
@@ -154,6 +154,10 @@ public class InternalEngine extends Engine {
this.versionMap.setManager(searcherManager);
// don't allow commits until we are done with recovering
allowCommits.compareAndSet(true, openMode != EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG);
+ if (engineConfig.getRefreshListeners() != null) {
+ searcherManager.addListener(engineConfig.getRefreshListeners());
+ engineConfig.getRefreshListeners().setTranslog(translog);
+ }
success = true;
} finally {
if (success == false) {
diff --git a/core/src/main/java/org/elasticsearch/index/engine/ShadowEngine.java b/core/src/main/java/org/elasticsearch/index/engine/ShadowEngine.java
index c30b2e9bf50..0a55803a5ec 100644
--- a/core/src/main/java/org/elasticsearch/index/engine/ShadowEngine.java
+++ b/core/src/main/java/org/elasticsearch/index/engine/ShadowEngine.java
@@ -30,7 +30,6 @@ import org.elasticsearch.common.lucene.Lucene;
import org.elasticsearch.common.lucene.index.ElasticsearchDirectoryReader;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.ReleasableLock;
-import org.elasticsearch.index.shard.TranslogRecoveryPerformer;
import org.elasticsearch.index.translog.Translog;
import java.io.IOException;
@@ -68,6 +67,9 @@ public class ShadowEngine extends Engine {
public ShadowEngine(EngineConfig engineConfig) {
super(engineConfig);
+ if (engineConfig.getRefreshListeners() != null) {
+ throw new IllegalArgumentException("ShadowEngine doesn't support RefreshListeners");
+ }
SearcherFactory searcherFactory = new EngineSearcherFactory(engineConfig);
final long nonexistentRetryTime = engineConfig.getIndexSettings().getSettings()
.getAsTime(NONEXISTENT_INDEX_RETRY_WAIT, DEFAULT_NONEXISTENT_INDEX_RETRY_WAIT)
diff --git a/core/src/main/java/org/elasticsearch/index/query/QueryBuilders.java b/core/src/main/java/org/elasticsearch/index/query/QueryBuilders.java
index d048f73381e..96585ce810f 100644
--- a/core/src/main/java/org/elasticsearch/index/query/QueryBuilders.java
+++ b/core/src/main/java/org/elasticsearch/index/query/QueryBuilders.java
@@ -27,7 +27,6 @@ import org.elasticsearch.common.geo.builders.ShapeBuilder;
import org.elasticsearch.index.query.MoreLikeThisQueryBuilder.Item;
import org.elasticsearch.index.query.functionscore.FunctionScoreQueryBuilder;
import org.elasticsearch.index.query.functionscore.ScoreFunctionBuilder;
-import org.elasticsearch.index.search.MatchQuery;
import org.elasticsearch.indices.TermsLookup;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptService;
@@ -57,7 +56,7 @@ public abstract class QueryBuilders {
* @param text The query text (to be analyzed).
*/
public static MatchQueryBuilder matchQuery(String name, Object text) {
- return new MatchQueryBuilder(name, text).type(MatchQuery.Type.BOOLEAN);
+ return new MatchQueryBuilder(name, text);
}
/**
@@ -86,8 +85,8 @@ public abstract class QueryBuilders {
* @param name The field name.
* @param text The query text (to be analyzed).
*/
- public static MatchQueryBuilder matchPhraseQuery(String name, Object text) {
- return new MatchQueryBuilder(name, text).type(MatchQuery.Type.PHRASE);
+ public static MatchPhraseQueryBuilder matchPhraseQuery(String name, Object text) {
+ return new MatchPhraseQueryBuilder(name, text);
}
/**
@@ -96,8 +95,8 @@ public abstract class QueryBuilders {
* @param name The field name.
* @param text The query text (to be analyzed).
*/
- public static MatchQueryBuilder matchPhrasePrefixQuery(String name, Object text) {
- return new MatchQueryBuilder(name, text).type(MatchQuery.Type.PHRASE_PREFIX);
+ public static MatchPhrasePrefixQueryBuilder matchPhrasePrefixQuery(String name, Object text) {
+ return new MatchPhrasePrefixQueryBuilder(name, text);
}
/**
diff --git a/core/src/main/java/org/elasticsearch/index/query/WrapperQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/WrapperQueryBuilder.java
index 4e1eb83272a..7e469d6bbe4 100644
--- a/core/src/main/java/org/elasticsearch/index/query/WrapperQueryBuilder.java
+++ b/core/src/main/java/org/elasticsearch/index/query/WrapperQueryBuilder.java
@@ -126,7 +126,7 @@ public class WrapperQueryBuilder extends AbstractQueryBuilder {
try {
- if (recoverFromLocalShards(mappingUpdateConsumer, startedShards)) {
+ final Set shards = IndexMetaData.selectShrinkShards(shardId().id(), sourceIndexService.getMetaData(),
+ indexMetaData.getNumberOfShards());
+ if (recoverFromLocalShards(mappingUpdateConsumer, startedShards.stream()
+ .filter((s) -> shards.contains(s.shardId())).collect(Collectors.toList()))) {
recoveryListener.onRecoveryDone(recoveryState);
}
} catch (Throwable t) {
@@ -1530,7 +1544,7 @@ public class IndexShard extends AbstractIndexShardComponent {
return new EngineConfig(openMode, shardId,
threadPool, indexSettings, warmer, store, deletionPolicy, indexSettings.getMergePolicy(),
mapperService.indexAnalyzer(), similarityService.similarity(mapperService), codecService, shardEventListener, translogRecoveryPerformer, indexCache.query(), cachingPolicy, translogConfig,
- IndexingMemoryController.SHARD_INACTIVE_TIME_SETTING.get(indexSettings.getSettings()));
+ IndexingMemoryController.SHARD_INACTIVE_TIME_SETTING.get(indexSettings.getSettings()), refreshListeners);
}
public Releasable acquirePrimaryOperationLock() {
@@ -1626,6 +1640,17 @@ public class IndexShard extends AbstractIndexShardComponent {
return false;
}
+ /**
+ * Build {@linkplain RefreshListeners} for this shard. Protected so {@linkplain ShadowIndexShard} can override it to return null.
+ */
+ protected RefreshListeners buildRefreshListeners() {
+ return new RefreshListeners(
+ indexSettings::getMaxRefreshListeners,
+ () -> refresh("too_many_listeners"),
+ threadPool.executor(ThreadPool.Names.LISTENER)::execute,
+ logger);
+ }
+
/**
* Simple struct encapsulating a shard failure
*
@@ -1651,14 +1676,26 @@ public class IndexShard extends AbstractIndexShardComponent {
}
/**
- * Returns true iff one or more changes to the engine are not visible to via the current searcher.
+ * Returns true iff one or more changes to the engine are not visible to via the current searcher *or* there are pending
+ * refresh listeners.
* Otherwise false.
*
* @throws EngineClosedException if the engine is already closed
* @throws AlreadyClosedException if the internal indexwriter in the engine is already closed
*/
public boolean isRefreshNeeded() {
- return getEngine().refreshNeeded();
+ return getEngine().refreshNeeded() || (refreshListeners != null && refreshListeners.refreshNeeded());
+ }
+
+ /**
+ * Add a listener for refreshes.
+ *
+ * @param location the location to listen for
+ * @param listener for the refresh. Called with true if registering the listener ran it out of slots and forced a refresh. Called with
+ * false otherwise.
+ */
+ public void addRefreshListener(Translog.Location location, Consumer listener) {
+ refreshListeners.addOrNotify(location, listener);
}
private class IndexShardRecoveryPerformer extends TranslogRecoveryPerformer {
diff --git a/core/src/main/java/org/elasticsearch/index/shard/RefreshListeners.java b/core/src/main/java/org/elasticsearch/index/shard/RefreshListeners.java
new file mode 100644
index 00000000000..ab3e334714a
--- /dev/null
+++ b/core/src/main/java/org/elasticsearch/index/shard/RefreshListeners.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.shard;
+
+import org.apache.lucene.search.ReferenceManager;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.index.translog.Translog;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.function.Consumer;
+import java.util.function.IntSupplier;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Allows for the registration of listeners that are called when a change becomes visible for search. This functionality is exposed from
+ * {@link IndexShard} but kept here so it can be tested without standing up the entire thing.
+ */
+public final class RefreshListeners implements ReferenceManager.RefreshListener {
+ private final IntSupplier getMaxRefreshListeners;
+ private final Runnable forceRefresh;
+ private final Executor listenerExecutor;
+ private final ESLogger logger;
+
+ /**
+ * List of refresh listeners. Defaults to null and built on demand because most refresh cycles won't need it. Entries are never removed
+ * from it, rather, it is nulled and rebuilt when needed again. The (hopefully) rare entries that didn't make the current refresh cycle
+ * are just added back to the new list. Both the reference and the contents are always modified while synchronized on {@code this}.
+ */
+ private volatile List>> refreshListeners = null;
+ /**
+ * The translog location that was last made visible by a refresh.
+ */
+ private volatile Translog.Location lastRefreshedLocation;
+
+ public RefreshListeners(IntSupplier getMaxRefreshListeners, Runnable forceRefresh, Executor listenerExecutor, ESLogger logger) {
+ this.getMaxRefreshListeners = getMaxRefreshListeners;
+ this.forceRefresh = forceRefresh;
+ this.listenerExecutor = listenerExecutor;
+ this.logger = logger;
+ }
+
+ /**
+ * Add a listener for refreshes, calling it immediately if the location is already visible. If this runs out of listener slots then it
+ * forces a refresh and calls the listener immediately as well.
+ *
+ * @param location the location to listen for
+ * @param listener for the refresh. Called with true if registering the listener ran it out of slots and forced a refresh. Called with
+ * false otherwise.
+ */
+ public void addOrNotify(Translog.Location location, Consumer listener) {
+ requireNonNull(listener, "listener cannot be null");
+ requireNonNull(location, "location cannot be null");
+
+ if (lastRefreshedLocation != null && lastRefreshedLocation.compareTo(location) >= 0) {
+ // Location already visible, just call the listener
+ listener.accept(false);
+ return;
+ }
+ synchronized (this) {
+ if (refreshListeners == null) {
+ refreshListeners = new ArrayList<>();
+ }
+ if (refreshListeners.size() < getMaxRefreshListeners.getAsInt()) {
+ // We have a free slot so register the listener
+ refreshListeners.add(new Tuple<>(location, listener));
+ return;
+ }
+ }
+ // No free slot so force a refresh and call the listener in this thread
+ forceRefresh.run();
+ listener.accept(true);
+ }
+
+ /**
+ * Returns true if there are pending listeners.
+ */
+ public boolean refreshNeeded() {
+ // No need to synchronize here because we're doing a single volatile read
+ return refreshListeners != null;
+ }
+
+ /**
+ * Setup the translog used to find the last refreshed location.
+ */
+ public void setTranslog(Translog translog) {
+ this.translog = translog;
+ }
+
+ // Implementation of ReferenceManager.RefreshListener that adapts Lucene's RefreshListener into Elasticsearch's refresh listeners.
+ private Translog translog;
+ /**
+ * Snapshot of the translog location before the current refresh if there is a refresh going on or null. Doesn't have to be volatile
+ * because when it is used by the refreshing thread.
+ */
+ private Translog.Location currentRefreshLocation;
+
+ @Override
+ public void beforeRefresh() throws IOException {
+ currentRefreshLocation = translog.getLastWriteLocation();
+ }
+
+ @Override
+ public void afterRefresh(boolean didRefresh) throws IOException {
+ /*
+ * We intentionally ignore didRefresh here because our timing is a little off. It'd be a useful flag if we knew everything that made
+ * it into the refresh, but the way we snapshot the translog position before the refresh, things can sneak into the refresh that we
+ * don't know about.
+ */
+ if (null == currentRefreshLocation) {
+ /*
+ * The translog had an empty last write location at the start of the refresh so we can't alert anyone to anything. This
+ * usually happens during recovery. The next refresh cycle out to pick up this refresh.
+ */
+ return;
+ }
+ // First check if we've actually moved forward. If not then just bail immediately.
+ assert lastRefreshedLocation == null || currentRefreshLocation.compareTo(lastRefreshedLocation) >= 0;
+ if (lastRefreshedLocation != null && currentRefreshLocation.compareTo(lastRefreshedLocation) == 0) {
+ return;
+ }
+ /*
+ * Set the lastRefreshedLocation so listeners that come in for locations before that will just execute inline without messing
+ * around with refreshListeners or synchronizing at all.
+ */
+ lastRefreshedLocation = currentRefreshLocation;
+ /*
+ * Grab the current refresh listeners and replace them with null while synchronized. Any listeners that come in after this won't be
+ * in the list we iterate over and very likely won't be candidates for refresh anyway because we've already moved the
+ * lastRefreshedLocation.
+ */
+ List>> candidates;
+ synchronized (this) {
+ candidates = refreshListeners;
+ // No listeners to check so just bail early
+ if (candidates == null) {
+ return;
+ }
+ refreshListeners = null;
+ }
+ // Iterate the list of listeners, copying the listeners to fire to one list and those to preserve to another list.
+ List