Merge branch 'master' into feature/multi_cluster_search

buildSrc/src/main/resources/checkstyle_suppressions.xml

@@ -311,7 +311,6 @@
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]AlreadyExpiredException.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]CompositeIndexEventListener.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]IndexSettings.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]IndexingSlowLog.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]MergePolicyConfig.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]SearchSlowLog.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]analysis[/\\]AnalysisRegistry.java" checks="LineLength" />
@@ -787,7 +786,6 @@
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]search[/\\]nested[/\\]LongNestedSortingTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]search[/\\]nested[/\\]NestedSortingTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]shard[/\\]IndexShardTests.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]shard[/\\]IndexingOperationListenerTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]shard[/\\]ShardPathTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]similarity[/\\]SimilarityTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]snapshots[/\\]blobstore[/\\]FileInfoTests.java" checks="LineLength" />

core/src/main/java/org/elasticsearch/common/network/NetworkModule.java

@@ -125,7 +125,8 @@ public final class NetworkModule {
             for (Map.Entry<String, Supplier<Transport>> entry : httpTransportFactory.entrySet()) {
                 registerTransport(entry.getKey(), entry.getValue());
             }
-            List<TransportInterceptor> transportInterceptors = plugin.getTransportInterceptors(namedWriteableRegistry);
+            List<TransportInterceptor> transportInterceptors = plugin.getTransportInterceptors(namedWriteableRegistry,
+                threadPool.getThreadContext());
             for (TransportInterceptor interceptor : transportInterceptors) {
                 registerTransportInterceptor(interceptor);
             }

core/src/main/java/org/elasticsearch/index/IndexingSlowLog.java

@@ -30,6 +30,7 @@ import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.shard.IndexingOperationListener;
+import org.elasticsearch.index.shard.ShardId;
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
@@ -76,13 +77,14 @@ public final class IndexingSlowLog implements IndexingOperationListener {
      * and everything else is interpreted as Elasticsearch interprets booleans
      * which is then converted to 0 for false and Integer.MAX_VALUE for true.
      */
-    public static final Setting<Integer> INDEX_INDEXING_SLOWLOG_MAX_SOURCE_CHARS_TO_LOG_SETTING = new Setting<>(INDEX_INDEXING_SLOWLOG_PREFIX + ".source", "1000", (value) -> {
-        try {
-            return Integer.parseInt(value, 10);
-        } catch (NumberFormatException e) {
-            return Booleans.parseBoolean(value, true) ? Integer.MAX_VALUE : 0;
-        }
-    }, Property.Dynamic, Property.IndexScope);
+    public static final Setting<Integer> INDEX_INDEXING_SLOWLOG_MAX_SOURCE_CHARS_TO_LOG_SETTING =
+            new Setting<>(INDEX_INDEXING_SLOWLOG_PREFIX + ".source", "1000", (value) -> {
+                try {
+                    return Integer.parseInt(value, 10);
+                } catch (NumberFormatException e) {
+                    return Booleans.parseBoolean(value, true) ? Integer.MAX_VALUE : 0;
+                }
+            }, Property.Dynamic, Property.IndexScope);
 
     IndexingSlowLog(IndexSettings indexSettings) {
         this.indexLogger = Loggers.getLogger(INDEX_INDEXING_SLOWLOG_PREFIX + ".index", indexSettings.getSettings());
@@ -90,17 +92,22 @@ public final class IndexingSlowLog implements IndexingOperationListener {
 
         indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_REFORMAT_SETTING, this::setReformat);
         this.reformat = indexSettings.getValue(INDEX_INDEXING_SLOWLOG_REFORMAT_SETTING);
-        indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_WARN_SETTING, this::setWarnThreshold);
+        indexSettings.getScopedSettings()
+                .addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_WARN_SETTING, this::setWarnThreshold);
         this.indexWarnThreshold = indexSettings.getValue(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_WARN_SETTING).nanos();
-        indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_INFO_SETTING, this::setInfoThreshold);
+        indexSettings.getScopedSettings()
+                .addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_INFO_SETTING, this::setInfoThreshold);
         this.indexInfoThreshold = indexSettings.getValue(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_INFO_SETTING).nanos();
-        indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_DEBUG_SETTING, this::setDebugThreshold);
+        indexSettings.getScopedSettings()
+                .addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_DEBUG_SETTING, this::setDebugThreshold);
         this.indexDebugThreshold = indexSettings.getValue(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_DEBUG_SETTING).nanos();
-        indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_TRACE_SETTING, this::setTraceThreshold);
+        indexSettings.getScopedSettings()
+                .addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_TRACE_SETTING, this::setTraceThreshold);
         this.indexTraceThreshold = indexSettings.getValue(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_TRACE_SETTING).nanos();
         indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_LEVEL_SETTING, this::setLevel);
         setLevel(indexSettings.getValue(INDEX_INDEXING_SLOWLOG_LEVEL_SETTING));
-        indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_MAX_SOURCE_CHARS_TO_LOG_SETTING, this::setMaxSourceCharsToLog);
+        indexSettings.getScopedSettings().addSettingsUpdateConsumer(INDEX_INDEXING_SLOWLOG_MAX_SOURCE_CHARS_TO_LOG_SETTING,
+                this::setMaxSourceCharsToLog);
         this.maxSourceCharsToLog = indexSettings.getValue(INDEX_INDEXING_SLOWLOG_MAX_SOURCE_CHARS_TO_LOG_SETTING);
     }
 
@@ -134,7 +141,7 @@ public final class IndexingSlowLog implements IndexingOperationListener {
     }
 
     @Override
-    public void postIndex(Engine.Index indexOperation, Engine.IndexResult result) {
+    public void postIndex(ShardId shardId, Engine.Index indexOperation, Engine.IndexResult result) {
         if (result.hasFailure() == false) {
             final ParsedDocument doc = indexOperation.parsedDoc();
             final long tookInNanos = result.getTook();
@@ -169,7 +176,8 @@ public final class IndexingSlowLog implements IndexingOperationListener {
         public String toString() {
             StringBuilder sb = new StringBuilder();
             sb.append(index).append(" ");
-            sb.append("took[").append(TimeValue.timeValueNanos(tookInNanos)).append("], took_millis[").append(TimeUnit.NANOSECONDS.toMillis(tookInNanos)).append("], ");
+            sb.append("took[").append(TimeValue.timeValueNanos(tookInNanos)).append("], ");
+            sb.append("took_millis[").append(TimeUnit.NANOSECONDS.toMillis(tookInNanos)).append("], ");
             sb.append("type[").append(doc.type()).append("], ");
             sb.append("id[").append(doc.id()).append("], ");
             if (doc.routing() == null) {

core/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -553,17 +553,17 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     private Engine.IndexResult index(Engine engine, Engine.Index index) {
         active.set(true);
         final Engine.IndexResult result;
-        index = indexingOperationListeners.preIndex(index);
+        index = indexingOperationListeners.preIndex(shardId, index);
         try {
             if (logger.isTraceEnabled()) {
                 logger.trace("index [{}][{}]{}", index.type(), index.id(), index.docs());
             }
             result = engine.index(index);
         } catch (Exception e) {
-            indexingOperationListeners.postIndex(index, e);
+            indexingOperationListeners.postIndex(shardId, index, e);
             throw e;
         }
-        indexingOperationListeners.postIndex(index, result);
+        indexingOperationListeners.postIndex(shardId, index, result);
         return result;
     }
 
@@ -602,17 +602,17 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     private Engine.DeleteResult delete(Engine engine, Engine.Delete delete) {
         active.set(true);
         final Engine.DeleteResult result;
-        delete = indexingOperationListeners.preDelete(delete);
+        delete = indexingOperationListeners.preDelete(shardId, delete);
         try {
             if (logger.isTraceEnabled()) {
                 logger.trace("delete [{}]", delete.uid().text());
             }
             result = engine.delete(delete);
         } catch (Exception e) {
-            indexingOperationListeners.postDelete(delete, e);
+            indexingOperationListeners.postDelete(shardId, delete, e);
             throw e;
         }
-        indexingOperationListeners.postDelete(delete, result);
+        indexingOperationListeners.postDelete(shardId, delete, result);
         return result;
     }
 

core/src/main/java/org/elasticsearch/index/shard/IndexingOperationListener.java

@@ -33,29 +33,29 @@ public interface IndexingOperationListener {
     /**
      * Called before the indexing occurs.
      */
-    default Engine.Index preIndex(Engine.Index operation) {
+    default Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
         return operation;
     }
 
     /**
      * Called after the indexing operation occurred. Note that this is
      * also called when indexing a document did not succeed due to document
-     * related failures. See {@link #postIndex(Engine.Index, Exception)}
+     * related failures. See {@link #postIndex(ShardId, Engine.Index, Exception)}
      * for engine level failures
      */
-    default void postIndex(Engine.Index index, Engine.IndexResult result) {}
+    default void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {}
 
     /**
      * Called after the indexing operation occurred with engine level exception.
-     * See {@link #postIndex(Engine.Index, Engine.IndexResult)} for document
+     * See {@link #postIndex(ShardId, Engine.Index, Engine.IndexResult)} for document
      * related failures
      */
-    default void postIndex(Engine.Index index, Exception ex) {}
+    default void postIndex(ShardId shardId, Engine.Index index, Exception ex) {}
 
     /**
      * Called before the delete occurs.
      */
-    default Engine.Delete preDelete(Engine.Delete delete) {
+    default Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
         return delete;
     }
 
@@ -63,17 +63,17 @@ public interface IndexingOperationListener {
     /**
      * Called after the delete operation occurred. Note that this is
      * also called when deleting a document did not succeed due to document
-     * related failures. See {@link #postDelete(Engine.Delete, Exception)}
+     * related failures. See {@link #postDelete(ShardId, Engine.Delete, Exception)}
      * for engine level failures
      */
-    default void postDelete(Engine.Delete delete, Engine.DeleteResult result) {}
+    default void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {}
 
     /**
      * Called after the delete operation occurred with engine level exception.
-     * See {@link #postDelete(Engine.Delete, Engine.DeleteResult)} for document
+     * See {@link #postDelete(ShardId, Engine.Delete, Engine.DeleteResult)} for document
      * related failures
      */
-    default void postDelete(Engine.Delete delete, Exception ex) {}
+    default void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {}
 
     /**
      * A Composite listener that multiplexes calls to each of the listeners methods.
@@ -88,11 +88,11 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public Engine.Index preIndex(Engine.Index operation) {
+        public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
             assert operation != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
-                    listener.preIndex(operation);
+                    listener.preIndex(shardId, operation);
                 } catch (Exception e) {
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("preIndex listener [{}] failed", listener), e);
                 }
@@ -101,11 +101,11 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public void postIndex(Engine.Index index, Engine.IndexResult result) {
+        public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
             assert index != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
-                    listener.postIndex(index, result);
+                    listener.postIndex(shardId, index, result);
                 } catch (Exception e) {
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("postIndex listener [{}] failed", listener), e);
                 }
@@ -113,11 +113,11 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public void postIndex(Engine.Index index, Exception ex) {
+        public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
             assert index != null && ex != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
-                    listener.postIndex(index, ex);
+                    listener.postIndex(shardId, index, ex);
                 } catch (Exception inner) {
                     inner.addSuppressed(ex);
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("postIndex listener [{}] failed", listener), inner);
@@ -126,11 +126,11 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public Engine.Delete preDelete(Engine.Delete delete) {
+        public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
             assert delete != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
-                    listener.preDelete(delete);
+                    listener.preDelete(shardId, delete);
                 } catch (Exception e) {
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("preDelete listener [{}] failed", listener), e);
                 }
@@ -139,11 +139,11 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+        public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
             assert delete != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
-                    listener.postDelete(delete, result);
+                    listener.postDelete(shardId, delete, result);
                 } catch (Exception e) {
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("postDelete listener [{}] failed", listener), e);
                 }
@@ -151,11 +151,11 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public void postDelete(Engine.Delete delete, Exception ex) {
+        public void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {
             assert delete != null && ex != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
-                    listener.postDelete(delete, ex);
+                    listener.postDelete(shardId, delete, ex);
                 } catch (Exception inner) {
                     inner.addSuppressed(ex);
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("postDelete listener [{}] failed", listener), inner);

core/src/main/java/org/elasticsearch/index/shard/InternalIndexingStats.java

@@ -65,7 +65,7 @@ final class InternalIndexingStats implements IndexingOperationListener {
     }
 
     @Override
-    public Engine.Index preIndex(Engine.Index operation) {
+    public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
         if (!operation.origin().isRecovery()) {
             totalStats.indexCurrent.inc();
             typeStats(operation.type()).indexCurrent.inc();
@@ -74,7 +74,7 @@ final class InternalIndexingStats implements IndexingOperationListener {
     }
 
     @Override
-    public void postIndex(Engine.Index index, Engine.IndexResult result) {
+    public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
         if (result.hasFailure() == false) {
             if (!index.origin().isRecovery()) {
                 long took = result.getTook();
@@ -85,12 +85,12 @@ final class InternalIndexingStats implements IndexingOperationListener {
                 typeStats.indexCurrent.dec();
             }
         } else {
-            postIndex(index, result.getFailure());
+            postIndex(shardId, index, result.getFailure());
         }
     }
 
     @Override
-    public void postIndex(Engine.Index index, Exception ex) {
+    public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
         if (!index.origin().isRecovery()) {
             totalStats.indexCurrent.dec();
             typeStats(index.type()).indexCurrent.dec();
@@ -100,7 +100,7 @@ final class InternalIndexingStats implements IndexingOperationListener {
     }
 
     @Override
-    public Engine.Delete preDelete(Engine.Delete delete) {
+    public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
         if (!delete.origin().isRecovery()) {
             totalStats.deleteCurrent.inc();
             typeStats(delete.type()).deleteCurrent.inc();
@@ -110,7 +110,7 @@ final class InternalIndexingStats implements IndexingOperationListener {
     }
 
     @Override
-    public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+    public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
         if (result.hasFailure() == false) {
             if (!delete.origin().isRecovery()) {
                 long took = result.getTook();
@@ -121,12 +121,12 @@ final class InternalIndexingStats implements IndexingOperationListener {
                 typeStats.deleteCurrent.dec();
             }
         } else {
-            postDelete(delete, result.getFailure());
+            postDelete(shardId, delete, result.getFailure());
         }
     }
 
     @Override
-    public void postDelete(Engine.Delete delete, Exception ex) {
+    public void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {
         if (!delete.origin().isRecovery()) {
             totalStats.deleteCurrent.dec();
             typeStats(delete.type()).deleteCurrent.dec();

core/src/main/java/org/elasticsearch/indices/IndexingMemoryController.java

@@ -34,6 +34,7 @@ import org.elasticsearch.index.engine.EngineClosedException;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.IndexShardState;
 import org.elasticsearch.index.shard.IndexingOperationListener;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.threadpool.ThreadPool.Cancellable;
 import org.elasticsearch.threadpool.ThreadPool.Names;
@@ -200,12 +201,12 @@ public class IndexingMemoryController extends AbstractComponent implements Index
     }
 
     @Override
-    public void postIndex(Engine.Index index, Engine.IndexResult result) {
+    public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
         recordOperationBytes(index, result);
     }
 
     @Override
-    public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+    public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
         recordOperationBytes(delete, result);
     }
 

core/src/main/java/org/elasticsearch/monitor/os/OsProbe.java

@@ -203,13 +203,10 @@ public class OsProbe {
         final Map<String, String> controllerMap = new HashMap<>();
         for (final String line : lines) {
             final Matcher matcher = CONTROL_GROUP_PATTERN.matcher(line);
-            // note that Matcher#matches must be invoked as
-            // matching is lazy; this can not happen in an assert
-            // as assertions might not be enabled
+            // Matcher#matches must be invoked as matching is lazy; this can not happen in an assert as assertions might not be enabled
             final boolean matches = matcher.matches();
             assert matches : line;
-            // at this point we have captured the subsystems and the
-            // control group
+            // at this point we have captured the subsystems and the control group
             final String[] controllers = matcher.group(1).split(",");
             for (final String controller : controllers) {
                 controllerMap.put(controller, matcher.group(2));
@@ -376,7 +373,7 @@ public class OsProbe {
      * @return {@code true} if the stats are available, otherwise {@code false}
      */
     @SuppressForbidden(reason = "access /proc/self/cgroup, /sys/fs/cgroup/cpu, and /sys/fs/cgroup/cpuacct")
-    protected boolean areCgroupStatsAvailable() {
+    boolean areCgroupStatsAvailable() {
         if (!Files.exists(PathUtils.get("/proc/self/cgroup"))) {
             return false;
         }
@@ -421,9 +418,7 @@ public class OsProbe {
                     cpuStat);
             }
         } catch (final IOException e) {
-            if (logger.isDebugEnabled()) {
-                logger.debug("error reading control group stats", e);
-            }
+            logger.debug("error reading control group stats", e);
             return null;
         }
     }

core/src/main/java/org/elasticsearch/plugins/NetworkPlugin.java

@@ -27,6 +27,7 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.http.HttpServerTransport;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
@@ -42,8 +43,13 @@ public interface NetworkPlugin {
     /**
      * Returns a list of {@link TransportInterceptor} instances that are used to intercept incoming and outgoing
      * transport (inter-node) requests. This must not return <code>null</code>
+     *
+     * @param namedWriteableRegistry registry of all named writeables registered
+     * @param threadContext a {@link ThreadContext} of the current nodes or clients {@link ThreadPool} that can be used to set additional
+     *                      headers in the interceptors
      */
-    default List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry) {
+    default List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
+                                                                ThreadContext threadContext) {
         return Collections.emptyList();
     }
 

core/src/test/java/org/elasticsearch/action/IndicesRequestIT.java

@@ -80,6 +80,7 @@ import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.plugins.NetworkPlugin;
 import org.elasticsearch.plugins.Plugin;
@@ -743,7 +744,8 @@ public class IndicesRequestIT extends ESIntegTestCase {
         public static class TestPlugin extends Plugin implements NetworkPlugin {
             public final InterceptingTransportService instance = new InterceptingTransportService();
             @Override
-            public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry) {
+            public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
+                                                                       ThreadContext threadContext) {
                 return Collections.singletonList(instance);
             }
         }

core/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java

@@ -34,6 +34,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.plugins.NetworkPlugin;
 import org.elasticsearch.plugins.Plugin;
@@ -128,7 +129,8 @@ public class TransportClientHeadersTests extends AbstractClientHeadersTestCase {
             private InternalTransportServiceInterceptor instance = new InternalTransportServiceInterceptor();
 
             @Override
-            public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry) {
+            public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
+                                                                       ThreadContext threadContext) {
                 return Collections.singletonList(new TransportInterceptor() {
                     @Override
                     public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(String action, String executor,

core/src/test/java/org/elasticsearch/common/network/NetworkModuleTests.java

@@ -27,6 +27,7 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.BoundTransportAddress;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.http.HttpInfo;
 import org.elasticsearch.http.HttpServerAdapter;
@@ -37,6 +38,7 @@ import org.elasticsearch.plugins.NetworkPlugin;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.cat.AbstractCatAction;
+import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportInterceptor;
@@ -47,9 +49,23 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
 public class NetworkModuleTests extends ModuleTestCase {
+    private ThreadPool threadPool;
+
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+        threadPool = new TestThreadPool(NetworkModuleTests.class.getName());
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        super.tearDown();
+        ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS);
+    }
 
     static class FakeHttpTransport extends AbstractLifecycleComponent implements HttpServerTransport {
         public FakeHttpTransport() {
@@ -233,7 +249,9 @@ public class NetworkModuleTests extends ModuleTestCase {
         };
         NetworkModule module = newNetworkModule(settings, false, new NetworkPlugin() {
                 @Override
-                public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry) {
+                public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
+                                                                           ThreadContext threadContext) {
+                    assertNotNull(threadContext);
                     return Collections.singletonList(interceptor);
                 }
             });
@@ -246,7 +264,9 @@ public class NetworkModuleTests extends ModuleTestCase {
         NullPointerException nullPointerException = expectThrows(NullPointerException.class, () -> {
             newNetworkModule(settings, false, new NetworkPlugin() {
                 @Override
-                public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry) {
+                public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
+                                                                           ThreadContext threadContext) {
+                    assertNotNull(threadContext);
                     return Collections.singletonList(null);
                 }
             });
@@ -256,6 +276,6 @@ public class NetworkModuleTests extends ModuleTestCase {
     }
 
     private NetworkModule newNetworkModule(Settings settings, boolean transportClient, NetworkPlugin... plugins) {
-        return new NetworkModule(settings, transportClient, Arrays.asList(plugins), null, null, null, null, xContentRegistry(), null);
+        return new NetworkModule(settings, transportClient, Arrays.asList(plugins), threadPool, null, null, null, xContentRegistry(), null);
     }
 }

core/src/test/java/org/elasticsearch/index/IndexModuleTests.java

@@ -232,7 +232,7 @@ public class IndexModuleTests extends ESTestCase {
         AtomicBoolean executed = new AtomicBoolean(false);
         IndexingOperationListener listener = new IndexingOperationListener() {
             @Override
-            public Engine.Index preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
                 executed.set(true);
                 return operation;
             }
@@ -248,8 +248,9 @@ public class IndexModuleTests extends ESTestCase {
         assertSame(listener, indexService.getIndexOperationListeners().get(1));
 
         Engine.Index index = new Engine.Index(new Term("_uid", "1"), null);
+        ShardId shardId = new ShardId(new Index("foo", "bar"), 0);
         for (IndexingOperationListener l : indexService.getIndexOperationListeners()) {
-            l.preIndex(index);
+            l.preIndex(shardId, index);
         }
         assertTrue(executed.get());
         indexService.close("simon says", false);

core/src/test/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -420,7 +420,7 @@ public class IndexShardIT extends ESSingleNodeTestCase {
         IndexingOperationListener listener = new IndexingOperationListener() {
 
             @Override
-            public void postIndex(Engine.Index index, Engine.IndexResult result) {
+            public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
                 try {
                     assertNotNull(shardRef.get());
                     // this is all IMC needs to do - check current memory and refresh
@@ -434,7 +434,7 @@ public class IndexShardIT extends ESSingleNodeTestCase {
 
 
             @Override
-            public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
                 try {
                     assertNotNull(shardRef.get());
                     // this is all IMC needs to do - check current memory and refresh

core/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -573,13 +573,13 @@ public class IndexShardTests extends IndexShardTestCase {
         shard.close("simon says", true);
         shard = reinitShard(shard, new IndexingOperationListener() {
             @Override
-            public Engine.Index preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
                 preIndex.incrementAndGet();
                 return operation;
             }
 
             @Override
-            public void postIndex(Engine.Index index, Engine.IndexResult result) {
+            public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
                 if (result.hasFailure() == false) {
                     if (result.isCreated()) {
                         postIndexCreate.incrementAndGet();
@@ -587,32 +587,32 @@ public class IndexShardTests extends IndexShardTestCase {
                         postIndexUpdate.incrementAndGet();
                     }
                 } else {
-                    postIndex(index, result.getFailure());
+                    postIndex(shardId, index, result.getFailure());
                 }
             }
 
             @Override
-            public void postIndex(Engine.Index index, Exception ex) {
+            public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
                 postIndexException.incrementAndGet();
             }
 
             @Override
-            public Engine.Delete preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
                 preDelete.incrementAndGet();
                 return delete;
             }
 
             @Override
-            public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
                 if (result.hasFailure() == false) {
                     postDelete.incrementAndGet();
                 } else {
-                    postDelete(delete, result.getFailure());
+                    postDelete(shardId, delete, result.getFailure());
                 }
             }
 
             @Override
-            public void postDelete(Engine.Delete delete, Exception ex) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {
                 postDeleteException.incrementAndGet();
 
             }
@@ -1144,24 +1144,24 @@ public class IndexShardTests extends IndexShardTestCase {
         final AtomicInteger postDelete = new AtomicInteger();
         IndexingOperationListener listener = new IndexingOperationListener() {
             @Override
-            public Engine.Index preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
                 preIndex.incrementAndGet();
                 return operation;
             }
 
             @Override
-            public void postIndex(Engine.Index index, Engine.IndexResult result) {
+            public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
                 postIndex.incrementAndGet();
             }
 
             @Override
-            public Engine.Delete preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
                 preDelete.incrementAndGet();
                 return delete;
             }
 
             @Override
-            public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
                 postDelete.incrementAndGet();
 
             }

core/src/test/java/org/elasticsearch/index/shard/IndexingOperationListenerTests.java

@@ -19,6 +19,7 @@
 package org.elasticsearch.index.shard;
 
 import org.apache.lucene.index.Term;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.seqno.SequenceNumbersService;
 import org.elasticsearch.test.ESTestCase;
@@ -29,6 +30,8 @@ import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.hamcrest.Matchers.is;
+
 public class IndexingOperationListenerTests extends ESTestCase{
 
     // this test also tests if calls are correct if one or more listeners throw exceptions
@@ -39,76 +42,83 @@ public class IndexingOperationListenerTests extends ESTestCase{
         AtomicInteger preDelete = new AtomicInteger();
         AtomicInteger postDelete = new AtomicInteger();
         AtomicInteger postDeleteException = new AtomicInteger();
+        ShardId randomShardId = new ShardId(new Index(randomAsciiOfLength(10), randomAsciiOfLength(10)), randomIntBetween(1, 10));
         IndexingOperationListener listener = new IndexingOperationListener() {
             @Override
-            public Engine.Index preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
+                assertThat(shardId, is(randomShardId));
                 preIndex.incrementAndGet();
                 return operation;
             }
 
             @Override
-            public void postIndex(Engine.Index index, Engine.IndexResult result) {
+            public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
+                assertThat(shardId, is(randomShardId));
                 if (result.hasFailure() == false) {
                     postIndex.incrementAndGet();
                 } else {
-                    postIndex(index, result.getFailure());
+                    postIndex(shardId, index, result.getFailure());
                 }
             }
 
             @Override
-            public void postIndex(Engine.Index index, Exception ex) {
+            public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
+                assertThat(shardId, is(randomShardId));
                 postIndexException.incrementAndGet();
             }
 
             @Override
-            public Engine.Delete preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
+                assertThat(shardId, is(randomShardId));
                 preDelete.incrementAndGet();
                 return delete;
             }
 
             @Override
-            public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
+                assertThat(shardId, is(randomShardId));
                 if (result.hasFailure() == false) {
                     postDelete.incrementAndGet();
                 } else {
-                    postDelete(delete, result.getFailure());
+                    postDelete(shardId, delete, result.getFailure());
                 }
             }
 
             @Override
-            public void postDelete(Engine.Delete delete, Exception ex) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {
+                assertThat(shardId, is(randomShardId));
                 postDeleteException.incrementAndGet();
             }
         };
 
         IndexingOperationListener throwingListener = new IndexingOperationListener() {
             @Override
-            public Engine.Index preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
                 throw new RuntimeException();
             }
 
             @Override
-            public void postIndex(Engine.Index index, Engine.IndexResult result) {
+            public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
                 throw new RuntimeException();
             }
 
             @Override
-            public void postIndex(Engine.Index index, Exception ex) {
+            public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
                 throw new RuntimeException();
             }
 
             @Override
-            public Engine.Delete preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
                 throw new RuntimeException();
             }
 
             @Override
-            public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResult result) {
                 throw new RuntimeException();
             }
 
             @Override
-            public void postDelete(Engine.Delete delete, Exception ex) {
+            public void postDelete(ShardId shardId, Engine.Delete delete, Exception ex) {
                 throw new RuntimeException();
             }
         };
@@ -120,10 +130,11 @@ public class IndexingOperationListenerTests extends ESTestCase{
             }
         }
         Collections.shuffle(indexingOperationListeners, random());
-        IndexingOperationListener.CompositeListener compositeListener = new IndexingOperationListener.CompositeListener(indexingOperationListeners, logger);
+        IndexingOperationListener.CompositeListener compositeListener =
+                new IndexingOperationListener.CompositeListener(indexingOperationListeners, logger);
         Engine.Delete delete = new Engine.Delete("test", "1", new Term("_uid", "1"));
         Engine.Index index = new Engine.Index(new Term("_uid", "1"), null);
-        compositeListener.postDelete(delete, new Engine.DeleteResult(1, SequenceNumbersService.UNASSIGNED_SEQ_NO, true));
+        compositeListener.postDelete(randomShardId, delete, new Engine.DeleteResult(1, SequenceNumbersService.UNASSIGNED_SEQ_NO, true));
         assertEquals(0, preIndex.get());
         assertEquals(0, postIndex.get());
         assertEquals(0, postIndexException.get());
@@ -131,7 +142,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
         assertEquals(2, postDelete.get());
         assertEquals(0, postDeleteException.get());
 
-        compositeListener.postDelete(delete, new RuntimeException());
+        compositeListener.postDelete(randomShardId, delete, new RuntimeException());
         assertEquals(0, preIndex.get());
         assertEquals(0, postIndex.get());
         assertEquals(0, postIndexException.get());
@@ -139,7 +150,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
         assertEquals(2, postDelete.get());
         assertEquals(2, postDeleteException.get());
 
-        compositeListener.preDelete(delete);
+        compositeListener.preDelete(randomShardId, delete);
         assertEquals(0, preIndex.get());
         assertEquals(0, postIndex.get());
         assertEquals(0, postIndexException.get());
@@ -147,7 +158,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
         assertEquals(2, postDelete.get());
         assertEquals(2, postDeleteException.get());
 
-        compositeListener.postIndex(index, new Engine.IndexResult(0, SequenceNumbersService.UNASSIGNED_SEQ_NO, false));
+        compositeListener.postIndex(randomShardId, index, new Engine.IndexResult(0, SequenceNumbersService.UNASSIGNED_SEQ_NO, false));
         assertEquals(0, preIndex.get());
         assertEquals(2, postIndex.get());
         assertEquals(0, postIndexException.get());
@@ -155,7 +166,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
         assertEquals(2, postDelete.get());
         assertEquals(2, postDeleteException.get());
 
-        compositeListener.postIndex(index, new RuntimeException());
+        compositeListener.postIndex(randomShardId, index, new RuntimeException());
         assertEquals(0, preIndex.get());
         assertEquals(2, postIndex.get());
         assertEquals(2, postIndexException.get());
@@ -163,7 +174,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
         assertEquals(2, postDelete.get());
         assertEquals(2, postDeleteException.get());
 
-        compositeListener.preIndex(index);
+        compositeListener.preIndex(randomShardId, index);
         assertEquals(2, preIndex.get());
         assertEquals(2, postIndex.get());
         assertEquals(2, postIndexException.get());

core/src/test/java/org/elasticsearch/monitor/os/OsProbeTests.java

@@ -195,7 +195,7 @@ public class OsProbeTests extends ESTestCase {
             }
 
             @Override
-            protected boolean areCgroupStatsAvailable() {
+            boolean areCgroupStatsAvailable() {
                 return areCgroupStatsAvailable;
             }
 

docs/reference/docs.asciidoc

@@ -16,6 +16,7 @@ detailed description of the following CRUD APIs:
 .Multi-document APIs
 * <<docs-multi-get>>
 * <<docs-bulk>>
+* <<docs-delete-by-query>>
 * <<docs-update-by-query>>
 * <<docs-reindex>>
 

docs/reference/docs/update-by-query.asciidoc

@@ -501,6 +501,7 @@ Which results in a sensible `total` like this one:
 
 Adding `slices` to `_update_by_query` just automates the manual process used in
 the section above, creating sub-requests which means it has some quirks:
+
 * You can see these requests in the
 <<docs-update-by-query-task-api,Tasks APIs>>. These sub-requests are "child"
 tasks of the task for the request with `slices`.

docs/reference/getting-started.asciidoc

@@ -551,7 +551,7 @@ POST /customer/external/_bulk?pretty
 
 Note above that for the delete action, there is no corresponding source document after it since deletes only require the ID of the document to be deleted.
 
-The bulk API executes all the actions sequentially and in order. If a single action fails for whatever reason, it will continue to process the remainder of the actions after it. When the bulk API returns, it will provide a status for each action (in the same order it was sent in) so that you can check if a specific action failed or not.
+The Bulk API does not fail due to failures in one of the actions. If a single action fails for whatever reason, it will continue to process the remainder of the actions after it. When the bulk API returns, it will provide a status for each action (in the same order it was sent in) so that you can check if a specific action failed or not.
 
 == Exploring Your Data
 

modules/reindex/src/test/java/org/elasticsearch/index/reindex/CancelTests.java

@@ -29,6 +29,7 @@ import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.Engine.Operation.Origin;
 import org.elasticsearch.index.shard.IndexingOperationListener;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.ingest.IngestTestPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.tasks.TaskInfo;
@@ -269,12 +270,12 @@ public class CancelTests extends ReindexTestCase {
     public static class BlockingOperationListener implements IndexingOperationListener {
 
         @Override
-        public Engine.Index preIndex(Engine.Index index) {
+        public Engine.Index preIndex(ShardId shardId, Engine.Index index) {
             return preCheck(index, index.type());
         }
 
         @Override
-        public Engine.Delete preDelete(Engine.Delete delete) {
+        public Engine.Delete preDelete(ShardId shardId, Engine.Delete delete) {
             return preCheck(delete, delete.type());
         }
 

test/framework/src/main/java/org/elasticsearch/transport/AssertingTransportInterceptor.java

@@ -22,6 +22,7 @@ import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Streamable;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.plugins.NetworkPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.tasks.Task;
@@ -51,7 +52,8 @@ public final class AssertingTransportInterceptor implements TransportInterceptor
         }
 
         @Override
-        public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry) {
+        public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
+                                                                   ThreadContext threadContext) {
             return Collections.singletonList(new AssertingTransportInterceptor(settings, namedWriteableRegistry));
         }
     }