diff options
| author | Simon Willnauer <simonw@apache.org> | 2015-12-23 14:46:54 +0100 |
|---|---|---|
| committer | Simon Willnauer <simonw@apache.org> | 2016-01-13 11:53:32 +0100 |
| commit | 574d1b35b35e2eb511cd42ab32d3d945b9852e47 (patch) | |
| tree | 4883c65a88a4fad8ba4c20d7992c7f42bacbe188 /core/src/main/java/org/elasticsearch | |
| parent | a2796b555f5c174f78303a2ce0c5a6439259f84b (diff) | |
Replace ContextAndHeaders with a ThreadPool based ThreadLocal implementation
ContextAndHeaders has a massive impact on the core infrastructure since it has to
be manually passed on to all relevant places across threads/network calls etc. For the same reason
it's also very error prone and easily forgotten on potentially relevant APIs.
The new ThreadContext is associated with a ThreadPool (node or transport client) and ensures that
headers and context registered on a current thread are inherited to new threads spawned, send across
the network to be deserialized on the receiver end as well as restored on the response handling thread
once the response is received.
Diffstat (limited to 'core/src/main/java/org/elasticsearch')
220 files changed, 872 insertions, 1266 deletions
diff --git a/core/src/main/java/org/elasticsearch/action/ActionRequest.java b/core/src/main/java/org/elasticsearch/action/ActionRequest.java index 45e7e76ca0..eedcff04ae 100644 --- a/core/src/main/java/org/elasticsearch/action/ActionRequest.java +++ b/core/src/main/java/org/elasticsearch/action/ActionRequest.java @@ -34,13 +34,6 @@ public abstract class ActionRequest<T extends ActionRequest> extends TransportRe super(); } - protected ActionRequest(ActionRequest request) { - super(request); - // this does not set the listenerThreaded API, if needed, its up to the caller to set it - // since most times, we actually want it to not be threaded... - // this.listenerThreaded = request.listenerThreaded(); - } - public abstract ActionRequestValidationException validate(); @Override diff --git a/core/src/main/java/org/elasticsearch/action/ActionRequestBuilder.java b/core/src/main/java/org/elasticsearch/action/ActionRequestBuilder.java index 9ad449f1f4..8cbc405daf 100644 --- a/core/src/main/java/org/elasticsearch/action/ActionRequestBuilder.java +++ b/core/src/main/java/org/elasticsearch/action/ActionRequestBuilder.java @@ -49,12 +49,6 @@ public abstract class ActionRequestBuilder<Request extends ActionRequest, Respon return this.request; } - @SuppressWarnings("unchecked") - public final RequestBuilder putHeader(String key, Object value) { - request.putHeader(key, value); - return (RequestBuilder) this; - } - public ListenableActionFuture<Response> execute() { PlainListenableActionFuture<Response> future = new PlainListenableActionFuture<>(threadPool); execute(future); diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java index 79adbafa9b..b5c9577aff 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java @@ -141,7 +141,7 @@ public class TransportClusterHealthAction extends TransportMasterNodeReadAction< } assert waitFor >= 0; - final ClusterStateObserver observer = new ClusterStateObserver(clusterService, logger); + final ClusterStateObserver observer = new ClusterStateObserver(clusterService, logger, threadPool.getThreadContext()); final ClusterState state = observer.observedState(); if (waitFor == 0 || request.timeout().millis() == 0) { listener.onResponse(getResponse(request, state, waitFor, request.timeout().millis() == 0)); diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java index f26177a0ce..c743a1d2a9 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java @@ -102,7 +102,7 @@ public class TransportNodesHotThreadsAction extends TransportNodesAction<NodesHo } NodeRequest(String nodeId, NodesHotThreadsRequest request) { - super(request, nodeId); + super(nodeId); this.request = request; } diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java index 3062148912..2a763910dd 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java @@ -96,7 +96,7 @@ public class TransportNodesInfoAction extends TransportNodesAction<NodesInfoRequ } NodeInfoRequest(String nodeId, NodesInfoRequest request) { - super(request, nodeId); + super(nodeId); this.request = request; } diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java index 1660a6de4a..8460eb59ba 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java @@ -96,7 +96,7 @@ public class TransportNodesStatsAction extends TransportNodesAction<NodesStatsRe } NodeStatsRequest(String nodeId, NodesStatsRequest request) { - super(request, nodeId); + super(nodeId); this.request = request; } diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java index 8c856beaf6..9800e1e6ae 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java @@ -20,7 +20,6 @@ package org.elasticsearch.action.admin.cluster.snapshots.status; import org.elasticsearch.ElasticsearchException; -import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.FailedNodeException; import org.elasticsearch.action.support.ActionFilters; import org.elasticsearch.action.support.nodes.BaseNodeRequest; @@ -146,8 +145,8 @@ public class TransportNodesSnapshotsStatus extends TransportNodesAction<Transpor public Request() { } - public Request(ActionRequest request, String[] nodesIds) { - super(request, nodesIds); + public Request(String[] nodesIds) { + super(nodesIds); } public Request snapshotIds(SnapshotId[] snapshotIds) { @@ -213,7 +212,7 @@ public class TransportNodesSnapshotsStatus extends TransportNodesAction<Transpor } NodeRequest(String nodeId, TransportNodesSnapshotsStatus.Request request) { - super(request, nodeId); + super(nodeId); snapshotIds = request.snapshotIds; } diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java index b5bb259409..fc19dd986d 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java @@ -110,7 +110,7 @@ public class TransportSnapshotsStatusAction extends TransportMasterNodeAction<Sn snapshotIds[i] = currentSnapshots.get(i).snapshotId(); } - TransportNodesSnapshotsStatus.Request nodesRequest = new TransportNodesSnapshotsStatus.Request(request, nodesIds.toArray(new String[nodesIds.size()])) + TransportNodesSnapshotsStatus.Request nodesRequest = new TransportNodesSnapshotsStatus.Request(nodesIds.toArray(new String[nodesIds.size()])) .snapshotIds(snapshotIds).timeout(request.masterNodeTimeout()); transportNodesSnapshotsStatus.execute(nodesRequest, new ActionListener<TransportNodesSnapshotsStatus.NodesSnapshotStatus>() { @Override diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java index 3e4880d0a6..3fc2f4b631 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java @@ -132,7 +132,7 @@ public class TransportClusterStatsAction extends TransportNodesAction<ClusterSta } ClusterStatsNodeRequest(String nodeId, ClusterStatsRequest request) { - super(request, nodeId); + super(nodeId); this.request = request; } diff --git a/core/src/main/java/org/elasticsearch/action/admin/cluster/validate/template/TransportRenderSearchTemplateAction.java b/core/src/main/java/org/elasticsearch/action/admin/cluster/validate/template/TransportRenderSearchTemplateAction.java index f2bfb18c43..0b4250ef74 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/cluster/validate/template/TransportRenderSearchTemplateAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/cluster/validate/template/TransportRenderSearchTemplateAction.java @@ -57,7 +57,7 @@ public class TransportRenderSearchTemplateAction extends HandledTransportAction< @Override protected void doRun() throws Exception { - ExecutableScript executable = scriptService.executable(request.template(), ScriptContext.Standard.SEARCH, request, Collections.emptyMap()); + ExecutableScript executable = scriptService.executable(request.template(), ScriptContext.Standard.SEARCH, Collections.emptyMap()); BytesReference processedTemplate = (BytesReference) executable.run(); RenderSearchTemplateResponse response = new RenderSearchTemplateResponse(); response.source(processedTemplate); diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexRequest.java b/core/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexRequest.java index ac0d5740d0..d1c753047a 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexRequest.java +++ b/core/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexRequest.java @@ -82,14 +82,6 @@ public class CreateIndexRequest extends AcknowledgedRequest<CreateIndexRequest> } /** - * Constructs a new request to create an index that was triggered by a different request, - * provided as an argument so that its headers and context can be copied to the new request. - */ - public CreateIndexRequest(ActionRequest request) { - super(request); - } - - /** * Constructs a new request to create an index with the specified name. */ public CreateIndexRequest(String index) { diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java index 0152254d22..7dc55c08fa 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java +++ b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java @@ -42,17 +42,6 @@ public class FlushRequest extends BroadcastRequest<FlushRequest> { private boolean force = false; private boolean waitIfOngoing = false; - public FlushRequest() { - } - - /** - * Copy constructor that creates a new flush request that is a copy of the one provided as an argument. - * The new request will inherit though headers and context from the original request that caused it. - */ - public FlushRequest(ActionRequest originalRequest) { - super(originalRequest); - } - /** * Constructs a new flush request against one or more indices. If nothing is provided, all indices will * be flushed. diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/ShardFlushRequest.java b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/ShardFlushRequest.java index ccf06be8bd..3a9ec89db5 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/ShardFlushRequest.java +++ b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/ShardFlushRequest.java @@ -31,7 +31,7 @@ public class ShardFlushRequest extends ReplicationRequest<ShardFlushRequest> { private FlushRequest request = new FlushRequest(); public ShardFlushRequest(FlushRequest request, ShardId shardId) { - super(request, shardId); + super(shardId); this.request = request; } diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/SyncedFlushRequest.java b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/SyncedFlushRequest.java index 59719fe887..2a14d66a76 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/SyncedFlushRequest.java +++ b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/SyncedFlushRequest.java @@ -36,17 +36,6 @@ import java.util.Arrays; */ public class SyncedFlushRequest extends BroadcastRequest<SyncedFlushRequest> { - public SyncedFlushRequest() { - } - - /** - * Copy constructor that creates a new synced flush request that is a copy of the one provided as an argument. - * The new request will inherit though headers and context from the original request that caused it. - */ - public SyncedFlushRequest(ActionRequest originalRequest) { - super(originalRequest); - } - /** * Constructs a new synced flush request against one or more indices. If nothing is provided, all indices will * be sync flushed. diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/GetFieldMappingsIndexRequest.java b/core/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/GetFieldMappingsIndexRequest.java index 59844430b3..149cba9a33 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/GetFieldMappingsIndexRequest.java +++ b/core/src/main/java/org/elasticsearch/action/admin/indices/mapping/get/GetFieldMappingsIndexRequest.java @@ -42,7 +42,6 @@ public class GetFieldMappingsIndexRequest extends SingleShardRequest<GetFieldMap } GetFieldMappingsIndexRequest(GetFieldMappingsRequest other, String index, boolean probablySingleFieldRequest) { - super(other); this.probablySingleFieldRequest = probablySingleFieldRequest; this.includeDefaults = other.includeDefaults(); this.types = other.types(); diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/RefreshRequest.java b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/RefreshRequest.java index ab9186c987..b5bce3c85c 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/RefreshRequest.java +++ b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/RefreshRequest.java @@ -33,17 +33,6 @@ import org.elasticsearch.action.support.broadcast.BroadcastRequest; */ public class RefreshRequest extends BroadcastRequest<RefreshRequest> { - public RefreshRequest() { - } - - /** - * Copy constructor that creates a new refresh request that is a copy of the one provided as an argument. - * The new request will inherit though headers and context from the original request that caused it. - */ - public RefreshRequest(ActionRequest originalRequest) { - super(originalRequest); - } - public RefreshRequest(String... indices) { super(indices); } diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java index a76b714b31..61aa4a86fc 100644 --- a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java +++ b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java @@ -54,7 +54,7 @@ public class TransportRefreshAction extends TransportBroadcastReplicationAction< @Override protected ReplicationRequest newShardRequest(RefreshRequest request, ShardId shardId) { - return new ReplicationRequest(request, shardId); + return new ReplicationRequest(shardId); } @Override diff --git a/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java b/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java index 02e0ea40d6..8ddb8c3298 100644 --- a/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java +++ b/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java @@ -69,14 +69,6 @@ public class BulkRequest extends ActionRequest<BulkRequest> implements Composite } /** - * Creates a bulk request caused by some other request, which is provided as an - * argument so that its headers and context can be copied to the new request - */ - public BulkRequest(ActionRequest request) { - super(request); - } - - /** * Adds a list of requests to be executed. Either index or delete requests. */ public BulkRequest add(ActionRequest... requests) { diff --git a/core/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java b/core/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java index 1edba16220..275e2819cf 100644 --- a/core/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java +++ b/core/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java @@ -41,7 +41,7 @@ public class BulkShardRequest extends ReplicationRequest<BulkShardRequest> { } BulkShardRequest(BulkRequest bulkRequest, ShardId shardId, boolean refresh, BulkItemRequest[] items) { - super(bulkRequest, shardId); + super(shardId); this.items = items; this.refresh = refresh; } diff --git a/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java b/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java index 7252993427..4750d9fb64 100644 --- a/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java +++ b/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java @@ -114,7 +114,7 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul for (Map.Entry<String, Set<String>> entry : indicesAndTypes.entrySet()) { final String index = entry.getKey(); if (autoCreateIndex.shouldAutoCreate(index, state)) { - CreateIndexRequest createIndexRequest = new CreateIndexRequest(bulkRequest); + CreateIndexRequest createIndexRequest = new CreateIndexRequest(); createIndexRequest.index(index); for (String type : entry.getValue()) { createIndexRequest.mapping(type); diff --git a/core/src/main/java/org/elasticsearch/action/delete/DeleteRequest.java b/core/src/main/java/org/elasticsearch/action/delete/DeleteRequest.java index ba63f33ee1..6c609eb9f3 100644 --- a/core/src/main/java/org/elasticsearch/action/delete/DeleteRequest.java +++ b/core/src/main/java/org/elasticsearch/action/delete/DeleteRequest.java @@ -92,7 +92,7 @@ public class DeleteRequest extends ReplicationRequest<DeleteRequest> implements * The new request will inherit though headers and context from the original request that caused it. */ public DeleteRequest(DeleteRequest request, ActionRequest originalRequest) { - super(request, originalRequest); + super(request); this.type = request.type(); this.id = request.id(); this.routing = request.routing(); @@ -102,14 +102,6 @@ public class DeleteRequest extends ReplicationRequest<DeleteRequest> implements this.versionType = request.versionType(); } - /** - * Creates a delete request caused by some other request, which is provided as an - * argument so that its headers and context can be copied to the new request - */ - public DeleteRequest(ActionRequest request) { - super(request); - } - @Override public ActionRequestValidationException validate() { ActionRequestValidationException validationException = super.validate(); diff --git a/core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java b/core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java index f80b1a2439..c235144c66 100644 --- a/core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java +++ b/core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java @@ -72,7 +72,7 @@ public class TransportDeleteAction extends TransportReplicationAction<DeleteRequ protected void doExecute(final DeleteRequest request, final ActionListener<DeleteResponse> listener) { ClusterState state = clusterService.state(); if (autoCreateIndex.shouldAutoCreate(request.index(), state)) { - createIndexAction.execute(new CreateIndexRequest(request).index(request.index()).cause("auto(delete api)").masterNodeTimeout(request.timeout()), new ActionListener<CreateIndexResponse>() { + createIndexAction.execute(new CreateIndexRequest().index(request.index()).cause("auto(delete api)").masterNodeTimeout(request.timeout()), new ActionListener<CreateIndexResponse>() { @Override public void onResponse(CreateIndexResponse result) { innerExecute(request, listener); diff --git a/core/src/main/java/org/elasticsearch/action/get/GetRequest.java b/core/src/main/java/org/elasticsearch/action/get/GetRequest.java index c6919e858d..1c83cbeb63 100644 --- a/core/src/main/java/org/elasticsearch/action/get/GetRequest.java +++ b/core/src/main/java/org/elasticsearch/action/get/GetRequest.java @@ -19,7 +19,6 @@ package org.elasticsearch.action.get; -import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.ActionRequestValidationException; import org.elasticsearch.action.RealtimeRequest; import org.elasticsearch.action.ValidateActions; @@ -72,8 +71,7 @@ public class GetRequest extends SingleShardRequest<GetRequest> implements Realti * Copy constructor that creates a new get request that is a copy of the one provided as an argument. * The new request will inherit though headers and context from the original request that caused it. */ - public GetRequest(GetRequest getRequest, ActionRequest originalRequest) { - super(originalRequest); + public GetRequest(GetRequest getRequest) { this.index = getRequest.index; this.type = getRequest.type; this.id = getRequest.id; @@ -99,14 +97,6 @@ public class GetRequest extends SingleShardRequest<GetRequest> implements Realti } /** - * Constructs a new get request starting from the provided request, meaning that it will - * inherit its headers and context, and against the specified index. - */ - public GetRequest(ActionRequest request, String index) { - super(request, index); - } - - /** * Constructs a new get request against the specified index with the type and id. * * @param index The index to get the document from diff --git a/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java b/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java index db3c0f72bf..f67e2b27b5 100644 --- a/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java +++ b/core/src/main/java/org/elasticsearch/action/get/MultiGetRequest.java @@ -266,18 +266,6 @@ public class MultiGetRequest extends ActionRequest<MultiGetRequest> implements I List<Item> items = new ArrayList<>(); - public MultiGetRequest() { - - } - - /** - * Creates a multi get request caused by some other request, which is provided as an - * argument so that its headers and context can be copied to the new request - */ - public MultiGetRequest(ActionRequest request) { - super(request); - } - public List<Item> getItems() { return this.items; } diff --git a/core/src/main/java/org/elasticsearch/action/get/MultiGetShardRequest.java b/core/src/main/java/org/elasticsearch/action/get/MultiGetShardRequest.java index 6715319bc3..9250204858 100644 --- a/core/src/main/java/org/elasticsearch/action/get/MultiGetShardRequest.java +++ b/core/src/main/java/org/elasticsearch/action/get/MultiGetShardRequest.java @@ -45,7 +45,7 @@ public class MultiGetShardRequest extends SingleShardRequest<MultiGetShardReques } MultiGetShardRequest(MultiGetRequest multiGetRequest, String index, int shardId) { - super(multiGetRequest, index); + super(index); this.shardId = shardId; locations = new IntArrayList(); items = new ArrayList<>(); diff --git a/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java b/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java index 9899a5493a..5057fdb829 100644 --- a/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java +++ b/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java @@ -21,7 +21,6 @@ package org.elasticsearch.action.index; import org.elasticsearch.ElasticsearchGenerationException; import org.elasticsearch.Version; -import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.ActionRequestValidationException; import org.elasticsearch.action.DocumentRequest; import org.elasticsearch.action.RoutingMissingException; @@ -159,19 +158,11 @@ public class IndexRequest extends ReplicationRequest<IndexRequest> implements Do } /** - * Creates an index request caused by some other request, which is provided as an - * argument so that its headers and context can be copied to the new request - */ - public IndexRequest(ActionRequest request) { - super(request); - } - - /** * Copy constructor that creates a new index request that is a copy of the one provided as an argument. * The new request will inherit though headers and context from the original request that caused it. */ - public IndexRequest(IndexRequest indexRequest, ActionRequest originalRequest) { - super(indexRequest, originalRequest); + public IndexRequest(IndexRequest indexRequest) { + super(indexRequest); this.type = indexRequest.type; this.id = indexRequest.id; this.routing = indexRequest.routing; diff --git a/core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java b/core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java index 620056ded4..4ae522d29c 100644 --- a/core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java +++ b/core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java @@ -88,7 +88,7 @@ public class TransportIndexAction extends TransportReplicationAction<IndexReques // if we don't have a master, we don't have metadata, that's fine, let it find a master using create index API ClusterState state = clusterService.state(); if (autoCreateIndex.shouldAutoCreate(request.index(), state)) { - CreateIndexRequest createIndexRequest = new CreateIndexRequest(request); + CreateIndexRequest createIndexRequest = new CreateIndexRequest(); createIndexRequest.index(request.index()); createIndexRequest.mapping(request.type()); createIndexRequest.cause("auto(index api)"); diff --git a/core/src/main/java/org/elasticsearch/action/percolate/PercolateRequest.java b/core/src/main/java/org/elasticsearch/action/percolate/PercolateRequest.java index 47f39cef0b..e69da6bf51 100644 --- a/core/src/main/java/org/elasticsearch/action/percolate/PercolateRequest.java +++ b/core/src/main/java/org/elasticsearch/action/percolate/PercolateRequest.java @@ -66,7 +66,6 @@ public class PercolateRequest extends BroadcastRequest<PercolateRequest> impleme } PercolateRequest(PercolateRequest request, BytesReference docSource) { - super(request); this.indices = request.indices(); this.documentType = request.documentType(); this.routing = request.routing(); @@ -274,7 +273,7 @@ public class PercolateRequest extends BroadcastRequest<PercolateRequest> impleme source = in.readBytesReference(); docSource = in.readBytesReference(); if (in.readBoolean()) { - getRequest = new GetRequest(null); + getRequest = new GetRequest(); getRequest.readFrom(in); } onlyCount = in.readBoolean(); diff --git a/core/src/main/java/org/elasticsearch/action/percolate/TransportMultiPercolateAction.java b/core/src/main/java/org/elasticsearch/action/percolate/TransportMultiPercolateAction.java index bf7b9e5307..987ca3c4cd 100644 --- a/core/src/main/java/org/elasticsearch/action/percolate/TransportMultiPercolateAction.java +++ b/core/src/main/java/org/elasticsearch/action/percolate/TransportMultiPercolateAction.java @@ -97,7 +97,7 @@ public class TransportMultiPercolateAction extends HandledTransportAction<MultiP } if (!existingDocsRequests.isEmpty()) { - final MultiGetRequest multiGetRequest = new MultiGetRequest(request); + final MultiGetRequest multiGetRequest = new MultiGetRequest(); for (GetRequest getRequest : existingDocsRequests) { multiGetRequest.add( new MultiGetRequest.Item(getRequest.index(), getRequest.type(), getRequest.id()) @@ -200,7 +200,7 @@ public class TransportMultiPercolateAction extends HandledTransportAction<MultiP ShardId shardId = shard.shardId(); TransportShardMultiPercolateAction.Request requests = requestsByShard.get(shardId); if (requests == null) { - requestsByShard.put(shardId, requests = new TransportShardMultiPercolateAction.Request(multiPercolateRequest, shardId.getIndex(), shardId.getId(), percolateRequest.preference())); + requestsByShard.put(shardId, requests = new TransportShardMultiPercolateAction.Request(shardId.getIndex(), shardId.getId(), percolateRequest.preference())); } logger.trace("Adding shard[{}] percolate request for item[{}]", shardId, slot); requests.add(new TransportShardMultiPercolateAction.Request.Item(slot, new PercolateShardRequest(shardId, percolateRequest))); diff --git a/core/src/main/java/org/elasticsearch/action/percolate/TransportPercolateAction.java b/core/src/main/java/org/elasticsearch/action/percolate/TransportPercolateAction.java index fdac839e14..bba024068c 100644 --- a/core/src/main/java/org/elasticsearch/action/percolate/TransportPercolateAction.java +++ b/core/src/main/java/org/elasticsearch/action/percolate/TransportPercolateAction.java @@ -74,7 +74,7 @@ public class TransportPercolateAction extends TransportBroadcastAction<Percolate request.startTime = System.currentTimeMillis(); if (request.getRequest() != null) { //create a new get request to make sure it has the same headers and context as the original percolate request - GetRequest getRequest = new GetRequest(request.getRequest(), request); + GetRequest getRequest = new GetRequest(request.getRequest()); getAction.execute(getRequest, new ActionListener<GetResponse>() { @Override public void onResponse(GetResponse getResponse) { @@ -150,7 +150,7 @@ public class TransportPercolateAction extends TransportBroadcastAction<Percolate } else { PercolatorService.ReduceResult result = null; try { - result = percolatorService.reduce(onlyCount, shardResults, request); + result = percolatorService.reduce(onlyCount, shardResults); } catch (IOException e) { throw new ElasticsearchException("error during reduce phase", e); } diff --git a/core/src/main/java/org/elasticsearch/action/percolate/TransportShardMultiPercolateAction.java b/core/src/main/java/org/elasticsearch/action/percolate/TransportShardMultiPercolateAction.java index 1d29e6c397..17b2fc1d61 100644 --- a/core/src/main/java/org/elasticsearch/action/percolate/TransportShardMultiPercolateAction.java +++ b/core/src/main/java/org/elasticsearch/action/percolate/TransportShardMultiPercolateAction.java @@ -118,8 +118,8 @@ public class TransportShardMultiPercolateAction extends TransportSingleShardActi public Request() { } - Request(MultiPercolateRequest multiPercolateRequest, String concreteIndex, int shardId, String preference) { - super(multiPercolateRequest, concreteIndex); + Request(String concreteIndex, int shardId, String preference) { + super(concreteIndex); this.shardId = shardId; this.preference = preference; this.items = new ArrayList<>(); diff --git a/core/src/main/java/org/elasticsearch/action/search/ClearScrollRequest.java b/core/src/main/java/org/elasticsearch/action/search/ClearScrollRequest.java index b390b77504..17343e8691 100644 --- a/core/src/main/java/org/elasticsearch/action/search/ClearScrollRequest.java +++ b/core/src/main/java/org/elasticsearch/action/search/ClearScrollRequest.java @@ -37,17 +37,6 @@ public class ClearScrollRequest extends ActionRequest<ClearScrollRequest> { private List<String> scrollIds; - public ClearScrollRequest() { - } - - /** - * Creates a clear scroll request caused by some other request, which is provided as an - * argument so that its headers and context can be copied to the new request - */ - public ClearScrollRequest(ActionRequest request) { - super(request); - } - public List<String> getScrollIds() { return scrollIds; } diff --git a/core/src/main/java/org/elasticsearch/action/search/SearchRequest.java b/core/src/main/java/org/elasticsearch/action/search/SearchRequest.java index 8014e4acb6..10a1ad2efc 100644 --- a/core/src/main/java/org/elasticsearch/action/search/SearchRequest.java +++ b/core/src/main/java/org/elasticsearch/action/search/SearchRequest.java @@ -80,8 +80,7 @@ public class SearchRequest extends ActionRequest<SearchRequest> implements Indic * Copy constructor that creates a new search request that is a copy of the one provided as an argument. * The new request will inherit though headers and context from the original request that caused it. */ - public SearchRequest(SearchRequest searchRequest, ActionRequest originalRequest) { - super(originalRequest); + public SearchRequest(SearchRequest searchRequest) { this.searchType = searchRequest.searchType; this.indices = searchRequest.indices; this.routing = searchRequest.routing; @@ -95,15 +94,6 @@ public class SearchRequest extends ActionRequest<SearchRequest> implements Indic } /** - * Constructs a new search request starting from the provided request, meaning that it will - * inherit its headers and context - */ - public SearchRequest(ActionRequest request) { - super(request); - this.source = new SearchSourceBuilder(); - } - - /** * Constructs a new search request against the indices. No indices provided here means that search * will run against all indices. */ diff --git a/core/src/main/java/org/elasticsearch/action/search/SearchScrollRequest.java b/core/src/main/java/org/elasticsearch/action/search/SearchScrollRequest.java index 537d61ac55..c1ff788dae 100644 --- a/core/src/main/java/org/elasticsearch/action/search/SearchScrollRequest.java +++ b/core/src/main/java/org/elasticsearch/action/search/SearchScrollRequest.java @@ -46,14 +46,6 @@ public class SearchScrollRequest extends ActionRequest<SearchScrollRequest> { this.scrollId = scrollId; } - /** - * Creates a scroll request caused by some other request, which is provided as an - * argument so that its headers and context can be copied to the new request - */ - public SearchScrollRequest(ActionRequest request) { - super(request); - } - @Override public ActionRequestValidationException validate() { ActionRequestValidationException validationException = null; diff --git a/core/src/main/java/org/elasticsearch/action/search/TransportMultiSearchAction.java b/core/src/main/java/org/elasticsearch/action/search/TransportMultiSearchAction.java index fd2b257e53..1849073563 100644 --- a/core/src/main/java/org/elasticsearch/action/search/TransportMultiSearchAction.java +++ b/core/src/main/java/org/elasticsearch/action/search/TransportMultiSearchAction.java @@ -59,7 +59,7 @@ public class TransportMultiSearchAction extends HandledTransportAction<MultiSear final AtomicInteger counter = new AtomicInteger(responses.length()); for (int i = 0; i < responses.length(); i++) { final int index = i; - SearchRequest searchRequest = new SearchRequest(request.requests().get(i), request); + SearchRequest searchRequest = new SearchRequest(request.requests().get(i)); searchAction.execute(searchRequest, new ActionListener<SearchResponse>() { @Override public void onResponse(SearchResponse searchResponse) { diff --git a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryAndFetchAction.java b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryAndFetchAction.java index 7244a1ff58..6d22264815 100644 --- a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryAndFetchAction.java +++ b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryAndFetchAction.java @@ -135,7 +135,7 @@ public class TransportSearchDfsQueryAndFetchAction extends TransportSearchTypeAc public void doRun() throws IOException { sortedShardList = searchPhaseController.sortDocs(true, queryFetchResults); final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryFetchResults, - queryFetchResults, request); + queryFetchResults); String scrollId = null; if (request.scroll() != null) { scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null); diff --git a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryThenFetchAction.java b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryThenFetchAction.java index faaf121554..31128cea96 100644 --- a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryThenFetchAction.java +++ b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchDfsQueryThenFetchAction.java @@ -211,7 +211,7 @@ public class TransportSearchDfsQueryThenFetchAction extends TransportSearchTypeA @Override public void doRun() throws IOException { final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryResults, - fetchResults, request); + fetchResults); String scrollId = null; if (request.scroll() != null) { scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null); diff --git a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryAndFetchAction.java b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryAndFetchAction.java index 3c4f5419f0..0e1e8db551 100644 --- a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryAndFetchAction.java +++ b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryAndFetchAction.java @@ -82,7 +82,7 @@ public class TransportSearchQueryAndFetchAction extends TransportSearchTypeActio boolean useScroll = request.scroll() != null; sortedShardList = searchPhaseController.sortDocs(useScroll, firstResults); final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, firstResults, - firstResults, request); + firstResults); String scrollId = null; if (request.scroll() != null) { scrollId = buildScrollId(request.searchType(), firstResults, null); diff --git a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryThenFetchAction.java b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryThenFetchAction.java index 1d8589e7a9..c63287d995 100644 --- a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryThenFetchAction.java +++ b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchQueryThenFetchAction.java @@ -146,7 +146,7 @@ public class TransportSearchQueryThenFetchAction extends TransportSearchTypeActi @Override public void doRun() throws IOException { final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, firstResults, - fetchResults, request); + fetchResults); String scrollId = null; if (request.scroll() != null) { scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults, null); diff --git a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryAndFetchAction.java b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryAndFetchAction.java index 2a953f9b73..b718baaa29 100644 --- a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryAndFetchAction.java +++ b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryAndFetchAction.java @@ -193,7 +193,7 @@ public class TransportSearchScrollQueryAndFetchAction extends AbstractComponent private void innerFinishHim() throws Exception { ScoreDoc[] sortedShardList = searchPhaseController.sortDocs(true, queryFetchResults); final InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryFetchResults, - queryFetchResults, request); + queryFetchResults); String scrollId = null; if (request.scroll() != null) { scrollId = request.scrollId(); diff --git a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryThenFetchAction.java b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryThenFetchAction.java index 8dd9c139c3..93a28b29aa 100644 --- a/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryThenFetchAction.java +++ b/core/src/main/java/org/elasticsearch/action/search/type/TransportSearchScrollQueryThenFetchAction.java @@ -208,7 +208,7 @@ public class TransportSearchScrollQueryThenFetchAction extends AbstractComponent IntArrayList docIds = entry.value; final QuerySearchResult querySearchResult = queryResults.get(entry.index); ScoreDoc lastEmittedDoc = lastEmittedDocPerShard[entry.index]; - ShardFetchRequest shardFetchRequest = new ShardFetchRequest(request, querySearchResult.id(), docIds, lastEmittedDoc); + ShardFetchRequest shardFetchRequest = new ShardFetchRequest(querySearchResult.id(), docIds, lastEmittedDoc); DiscoveryNode node = nodes.get(querySearchResult.shardTarget().nodeId()); searchService.sendExecuteFetchScroll(node, shardFetchRequest, new ActionListener<FetchSearchResult>() { @Override @@ -243,7 +243,7 @@ public class TransportSearchScrollQueryThenFetchAction extends AbstractComponent } private void innerFinishHim() { - InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryResults, fetchResults, request); + InternalSearchResponse internalResponse = searchPhaseController.merge(sortedShardList, queryResults, fetchResults); String scrollId = null; if (request.scroll() != null) { scrollId = request.scrollId(); diff --git a/core/src/main/java/org/elasticsearch/action/suggest/TransportSuggestAction.java b/core/src/main/java/org/elasticsearch/action/suggest/TransportSuggestAction.java index 6bc62cf83b..424d1b6a48 100644 --- a/core/src/main/java/org/elasticsearch/action/suggest/TransportSuggestAction.java +++ b/core/src/main/java/org/elasticsearch/action/suggest/TransportSuggestAction.java @@ -143,7 +143,7 @@ public class TransportSuggestAction extends TransportBroadcastAction<SuggestRequ throw new IllegalArgumentException("suggest content missing"); } final SuggestionSearchContext context = suggestPhase.parseElement().parseInternal(parser, indexService.mapperService(), - indexService.fieldData(), request.shardId().getIndex(), request.shardId().id(), request); + indexService.fieldData(), request.shardId().getIndex(), request.shardId().id()); final Suggest result = suggestPhase.execute(context, searcher.searcher()); return new ShardSuggestResponse(request.shardId(), result); } diff --git a/core/src/main/java/org/elasticsearch/action/support/ChildTaskRequest.java b/core/src/main/java/org/elasticsearch/action/support/ChildTaskRequest.java index c231028f0b..0483ec66e4 100644 --- a/core/src/main/java/org/elasticsearch/action/support/ChildTaskRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/ChildTaskRequest.java @@ -38,11 +38,6 @@ public class ChildTaskRequest extends TransportRequest { private long parentTaskId; protected ChildTaskRequest() { - - } - - protected ChildTaskRequest(TransportRequest parentTaskRequest) { - super(parentTaskRequest); } public void setParentTask(String parentTaskNode, long parentTaskId) { diff --git a/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastRequest.java b/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastRequest.java index bd070bdc88..b52944e5c8 100644 --- a/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastRequest.java @@ -37,11 +37,6 @@ public class BroadcastRequest<T extends BroadcastRequest> extends ActionRequest< private IndicesOptions indicesOptions = IndicesOptions.strictExpandOpenAndForbidClosed(); public BroadcastRequest() { - - } - - protected BroadcastRequest(ActionRequest originalRequest) { - super(originalRequest); } protected BroadcastRequest(String[] indices) { diff --git a/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java b/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java index 8e22a90b9d..921724e657 100644 --- a/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java @@ -42,7 +42,6 @@ public abstract class BroadcastShardRequest extends TransportRequest implements } protected BroadcastShardRequest(ShardId shardId, BroadcastRequest request) { - super(request); this.shardId = shardId; this.originalIndices = new OriginalIndices(request); } diff --git a/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java b/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java index e8f4a0d83c..30aeb86e13 100644 --- a/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java +++ b/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java @@ -433,7 +433,6 @@ public abstract class TransportBroadcastByNodeAction<Request extends BroadcastRe } public NodeRequest(String nodeId, Request request, List<ShardRouting> shards) { - super(request); this.indicesLevelRequest = request; this.shards = shards; this.nodeId = nodeId; diff --git a/core/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java b/core/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java index 156016e959..d8b1e104fa 100644 --- a/core/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java @@ -42,10 +42,6 @@ public abstract class AcknowledgedRequest<T extends MasterNodeRequest> extends M protected AcknowledgedRequest() { } - protected AcknowledgedRequest(ActionRequest request) { - super(request); - } - /** * Allows to set the timeout * @param timeout timeout as a string (e.g. 1s) diff --git a/core/src/main/java/org/elasticsearch/action/support/master/MasterNodeRequest.java b/core/src/main/java/org/elasticsearch/action/support/master/MasterNodeRequest.java index d791354d80..500c618467 100644 --- a/core/src/main/java/org/elasticsearch/action/support/master/MasterNodeRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/master/MasterNodeRequest.java @@ -36,11 +36,6 @@ public abstract class MasterNodeRequest<T extends MasterNodeRequest> extends Act protected TimeValue masterNodeTimeout = DEFAULT_MASTER_NODE_TIMEOUT; protected MasterNodeRequest() { - - } - - protected MasterNodeRequest(ActionRequest request) { - super(request); } /** diff --git a/core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java b/core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java index 60064a5ff4..d35bf300cc 100644 --- a/core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java +++ b/core/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java @@ -121,7 +121,7 @@ public abstract class TransportMasterNodeAction<Request extends MasterNodeReques } public void start() { - this.observer = new ClusterStateObserver(clusterService, request.masterNodeTimeout(), logger); + this.observer = new ClusterStateObserver(clusterService, request.masterNodeTimeout(), logger, threadPool.getThreadContext()); doStart(); } diff --git a/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodeRequest.java b/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodeRequest.java index 9371605e8a..9631fe659d 100644 --- a/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodeRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodeRequest.java @@ -36,8 +36,7 @@ public abstract class BaseNodeRequest extends ChildTaskRequest { } - protected BaseNodeRequest(BaseNodesRequest request, String nodeId) { - super(request); + protected BaseNodeRequest(String nodeId) { this.nodeId = nodeId; } diff --git a/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodesRequest.java b/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodesRequest.java index 804e2b0d1b..b9617c168a 100644 --- a/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodesRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/nodes/BaseNodesRequest.java @@ -43,11 +43,6 @@ public abstract class BaseNodesRequest<T extends BaseNodesRequest> extends Actio } - protected BaseNodesRequest(ActionRequest request, String... nodesIds) { - super(request); - this.nodesIds = nodesIds; - } - protected BaseNodesRequest(String... nodesIds) { this.nodesIds = nodesIds; } diff --git a/core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java b/core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java index adbe199824..61d2a5ab89 100644 --- a/core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java @@ -58,35 +58,20 @@ public class ReplicationRequest<T extends ReplicationRequest> extends ActionRequ } - /** - * Creates a new request that inherits headers and context from the request provided as argument. - */ - public ReplicationRequest(ActionRequest request) { - super(request); - } /** * Creates a new request with resolved shard id */ - public ReplicationRequest(ActionRequest request, ShardId shardId) { - super(request); + public ReplicationRequest(ShardId shardId) { this.index = shardId.getIndex(); this.shardId = shardId; } /** * Copy constructor that creates a new request that is a copy of the one provided as an argument. - */ - protected ReplicationRequest(T request) { - this(request, request); - } - - /** - * Copy constructor that creates a new request that is a copy of the one provided as an argument. * The new request will inherit though headers and context from the original request that caused it. */ - protected ReplicationRequest(T request, ActionRequest originalRequest) { - super(originalRequest); + protected ReplicationRequest(T request) { this.timeout = request.timeout(); this.index = request.index(); this.consistencyLevel = request.consistencyLevel(); diff --git a/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java b/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java index 0014404057..3c3ed714b5 100644 --- a/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java +++ b/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java @@ -52,6 +52,7 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.util.concurrent.AbstractRunnable; import org.elasticsearch.common.util.concurrent.ConcurrentCollections; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.index.IndexService; import org.elasticsearch.index.engine.VersionConflictEngineException; import org.elasticsearch.index.shard.IndexShard; @@ -302,7 +303,7 @@ public abstract class TransportReplicationAction<Request extends ReplicationRequ private final TransportChannel channel; // important: we pass null as a timeout as failing a replica is // something we want to avoid at all costs - private final ClusterStateObserver observer = new ClusterStateObserver(clusterService, null, logger); + private final ClusterStateObserver observer = new ClusterStateObserver(clusterService, null, logger, threadPool.getThreadContext()); AsyncReplicaAction(ReplicaRequest request, TransportChannel channel) { this.request = request; @@ -313,9 +314,12 @@ public abstract class TransportReplicationAction<Request extends ReplicationRequ public void onFailure(Throwable t) { if (t instanceof RetryOnReplicaException) { logger.trace("Retrying operation on replica, action [{}], request [{}]", t, transportReplicaAction, request); + final ThreadContext threadContext = threadPool.getThreadContext(); + final ThreadContext.StoredContext context = threadPool.getThreadContext().newStoredContext(); observer.waitForNextChange(new ClusterStateObserver.Listener() { @Override public void onNewClusterState(ClusterState state) { + context.close(); // Forking a thread on local node via transport service so that custom transport service have an // opportunity to execute custom logic before the replica operation begins String extraMessage = "action [" + transportReplicaAction + "], request[" + request + "]"; @@ -411,7 +415,7 @@ public abstract class TransportReplicationAction<Request extends ReplicationRequ ReroutePhase(Request request, ActionListener<Response> listener) { this.request = request; this.listener = listener; - this.observer = new ClusterStateObserver(clusterService, request.timeout(), logger); + this.observer = new ClusterStateObserver(clusterService, request.timeout(), logger, threadPool.getThreadContext()); } @Override @@ -515,9 +519,12 @@ public abstract class TransportReplicationAction<Request extends ReplicationRequ finishAsFailed(failure); return; } + final ThreadContext threadContext = threadPool.getThreadContext(); + final ThreadContext.StoredContext context = threadPool.getThreadContext().newStoredContext(); observer.waitForNextChange(new ClusterStateObserver.Listener() { @Override public void onNewClusterState(ClusterState state) { + context.close(); run(); } @@ -528,6 +535,7 @@ public abstract class TransportReplicationAction<Request extends ReplicationRequ @Override public void onTimeout(TimeValue timeout) { + context.close(); // Try one more time... run(); } diff --git a/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java b/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java index 50cdd0ebeb..833fa3a4db 100644 --- a/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java +++ b/core/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java @@ -124,7 +124,7 @@ public abstract class TransportInstanceSingleOperationAction<Request extends Ins } public void start() { - this.observer = new ClusterStateObserver(clusterService, request.timeout(), logger); + this.observer = new ClusterStateObserver(clusterService, request.timeout(), logger, threadPool.getThreadContext()); doStart(); } diff --git a/core/src/main/java/org/elasticsearch/action/support/single/shard/SingleShardRequest.java b/core/src/main/java/org/elasticsearch/action/support/single/shard/SingleShardRequest.java index 5d3484fdf9..4aa1fb5568 100644 --- a/core/src/main/java/org/elasticsearch/action/support/single/shard/SingleShardRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/single/shard/SingleShardRequest.java @@ -52,16 +52,7 @@ public abstract class SingleShardRequest<T extends SingleShardRequest> extends A public SingleShardRequest() { } - protected SingleShardRequest(String index) { - this.index = index; - } - - protected SingleShardRequest(ActionRequest request) { - super(request); - } - - protected SingleShardRequest(ActionRequest request, String index) { - super(request); + public SingleShardRequest(String index) { this.index = index; } diff --git a/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java b/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java index a1e485bb64..ea6698d439 100644 --- a/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java +++ b/core/src/main/java/org/elasticsearch/action/support/tasks/BaseTasksRequest.java @@ -65,15 +65,6 @@ public class BaseTasksRequest<T extends BaseTasksRequest> extends ActionRequest< * Get information about tasks from nodes based on the nodes ids specified. * If none are passed, information for all nodes will be returned. */ - public BaseTasksRequest(ActionRequest request, String... nodesIds) { - super(request); - this.nodesIds = nodesIds; - } - - /** - * Get information about tasks from nodes based on the nodes ids specified. - * If none are passed, information for all nodes will be returned. - */ public BaseTasksRequest(String... nodesIds) { this.nodesIds = nodesIds; } diff --git a/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java b/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java index 42be7e4eef..d2ce298565 100644 --- a/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java +++ b/core/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java @@ -291,7 +291,7 @@ public abstract class TransportTasksAction< } protected NodeTaskRequest(TasksRequest tasksRequest) { - super(tasksRequest); + super(); this.tasksRequest = tasksRequest; } diff --git a/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsShardRequest.java b/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsShardRequest.java index 5f541b02c7..6356c55499 100644 --- a/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsShardRequest.java +++ b/core/src/main/java/org/elasticsearch/action/termvectors/MultiTermVectorsShardRequest.java @@ -41,8 +41,8 @@ public class MultiTermVectorsShardRequest extends SingleShardRequest<MultiTermVe } - MultiTermVectorsShardRequest(MultiTermVectorsRequest request, String index, int shardId) { - super(request, index); + MultiTermVectorsShardRequest(String index, int shardId) { + super(index); this.shardId = shardId; locations = new IntArrayList(); requests = new ArrayList<>(); diff --git a/core/src/main/java/org/elasticsearch/action/termvectors/TransportMultiTermVectorsAction.java b/core/src/main/java/org/elasticsearch/action/termvectors/TransportMultiTermVectorsAction.java index 3943d2e6a6..535d89c101 100644 --- a/core/src/main/java/org/elasticsearch/action/termvectors/TransportMultiTermVectorsAction.java +++ b/core/src/main/java/org/elasticsearch/action/termvectors/TransportMultiTermVectorsAction.java @@ -82,7 +82,7 @@ public class TransportMultiTermVectorsAction extends HandledTransportAction<Mult termVectorsRequest.id(), termVectorsRequest.routing()); MultiTermVectorsShardRequest shardRequest = shardRequests.get(shardId); if (shardRequest == null) { - shardRequest = new MultiTermVectorsShardRequest(request, shardId.index().name(), shardId.id()); + shardRequest = new MultiTermVectorsShardRequest(shardId.index().name(), shardId.id()); shardRequest.preference(request.preference); shardRequests.put(shardId, shardRequest); } diff --git a/core/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java b/core/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java index 9ba1f2d1ea..b60403b00b 100644 --- a/core/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java +++ b/core/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java @@ -113,7 +113,7 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio protected void doExecute(final UpdateRequest request, final ActionListener<UpdateResponse> listener) { // if we don't have a master, we don't have metadata, that's fine, let it find a master using create index API if (autoCreateIndex.shouldAutoCreate(request.index(), clusterService.state())) { - createIndexAction.execute(new CreateIndexRequest(request).index(request.index()).cause("auto(update api)").masterNodeTimeout(request.timeout()), new ActionListener<CreateIndexResponse>() { + createIndexAction.execute(new CreateIndexRequest().index(request.index()).cause("auto(update api)").masterNodeTimeout(request.timeout()), new ActionListener<CreateIndexResponse>() { @Override public void onResponse(CreateIndexResponse result) { innerExecute(request, listener); @@ -164,12 +164,12 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio } protected void shardOperation(final UpdateRequest request, final ActionListener<UpdateResponse> listener, final int retryCount) { - IndexService indexService = indicesService.indexServiceSafe(request.concreteIndex()); - IndexShard indexShard = indexService.getShard(request.shardId()); + final IndexService indexService = indicesService.indexServiceSafe(request.concreteIndex()); + final IndexShard indexShard = indexService.getShard(request.shardId()); final UpdateHelper.Result result = updateHelper.prepare(request, indexShard); switch (result.operation()) { case UPSERT: - IndexRequest upsertRequest = new IndexRequest(result.action(), request); + IndexRequest upsertRequest = new IndexRequest((IndexRequest)result.action()); // we fetch it from the index request so we don't generate the bytes twice, its already done in the index request final BytesReference upsertSourceBytes = upsertRequest.source(); indexAction.execute(upsertRequest, new ActionListener<IndexResponse>() { @@ -206,7 +206,7 @@ public class TransportUpdateAction extends TransportInstanceSingleOperationActio }); break; case INDEX: - IndexRequest indexRequest = new IndexRequest(result.action(), request); + IndexRequest indexRequest = new IndexRequest((IndexRequest)result.action()); // we fetch it from the index request so we don't generate the bytes twice, its already done in the index request final BytesReference indexSourceBytes = indexRequest.source(); indexAction.execute(indexRequest, new ActionListener<IndexResponse>() { diff --git a/core/src/main/java/org/elasticsearch/action/update/UpdateHelper.java b/core/src/main/java/org/elasticsearch/action/update/UpdateHelper.java index d28ba2986e..48cf8a294c 100644 --- a/core/src/main/java/org/elasticsearch/action/update/UpdateHelper.java +++ b/core/src/main/java/org/elasticsearch/action/update/UpdateHelper.java @@ -44,6 +44,7 @@ import org.elasticsearch.index.mapper.internal.TimestampFieldMapper; import org.elasticsearch.index.shard.IndexShard; import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.script.ExecutableScript; +import org.elasticsearch.script.Script; import org.elasticsearch.script.ScriptContext; import org.elasticsearch.script.ScriptService; import org.elasticsearch.search.fetch.source.FetchSourceContext; @@ -99,7 +100,7 @@ public class UpdateHelper extends AbstractComponent { // Tell the script that this is a create and not an update ctx.put("op", "create"); ctx.put("_source", upsertDoc); - ctx = executeScript(request, ctx); + ctx = executeScript(request.script, ctx); //Allow the script to set TTL using ctx._ttl if (ttl == null) { ttl = getTTLFromScriptContext(ctx); @@ -193,7 +194,7 @@ public class UpdateHelper extends AbstractComponent { ctx.put("_ttl", originalTtl); ctx.put("_source", sourceAndContent.v2()); - ctx = executeScript(request, ctx); + ctx = executeScript(request.script, ctx); operation = (String) ctx.get("op"); @@ -243,14 +244,14 @@ public class UpdateHelper extends AbstractComponent { } } - private Map<String, Object> executeScript(UpdateRequest request, Map<String, Object> ctx) { + private Map<String, Object> executeScript(Script script, Map<String, Object> ctx) { try { if (scriptService != null) { - ExecutableScript script = scriptService.executable(request.script, ScriptContext.Standard.UPDATE, request, Collections.emptyMap()); - script.setNextVar("ctx", ctx); - script.run(); + ExecutableScript executableScript = scriptService.executable(script, ScriptContext.Standard.UPDATE, Collections.emptyMap()); + executableScript.setNextVar("ctx", ctx); + executableScript.run(); // we need to unwrap the ctx... - ctx = (Map<String, Object>) script.unwrap(ctx); + ctx = (Map<String, Object>) executableScript.unwrap(ctx); } } catch (Exception e) { throw new IllegalArgumentException("failed to execute script", e); diff --git a/core/src/main/java/org/elasticsearch/client/Client.java b/core/src/main/java/org/elasticsearch/client/Client.java index e7461dabfe..e1bcaf3ef5 100644 --- a/core/src/main/java/org/elasticsearch/client/Client.java +++ b/core/src/main/java/org/elasticsearch/client/Client.java @@ -19,8 +19,12 @@ package org.elasticsearch.client; +import org.elasticsearch.action.Action; import org.elasticsearch.action.ActionFuture; import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.ActionRequest; +import org.elasticsearch.action.ActionRequestBuilder; +import org.elasticsearch.action.ActionResponse; import org.elasticsearch.action.bulk.BulkRequest; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; @@ -80,11 +84,12 @@ import org.elasticsearch.action.termvectors.TermVectorsResponse; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateRequestBuilder; import org.elasticsearch.action.update.UpdateResponse; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.lease.Releasable; import org.elasticsearch.common.settings.Settings; +import java.util.Map; + /** * A client provides a one stop interface for performing actions/operations against the cluster. * <p> @@ -597,5 +602,5 @@ public interface Client extends ElasticsearchClient, Releasable { */ Settings settings(); - Headers headers(); + Client filterWithHeader(Map<String, String> headers); } diff --git a/core/src/main/java/org/elasticsearch/client/FilterClient.java b/core/src/main/java/org/elasticsearch/client/FilterClient.java index 06d81f0c9d..be7160fca0 100644 --- a/core/src/main/java/org/elasticsearch/client/FilterClient.java +++ b/core/src/main/java/org/elasticsearch/client/FilterClient.java @@ -42,7 +42,7 @@ public abstract class FilterClient extends AbstractClient { * @see #in() */ public FilterClient(Client in) { - super(in.settings(), in.threadPool(), in.headers()); + super(in.settings(), in.threadPool()); this.in = in; } diff --git a/core/src/main/java/org/elasticsearch/client/node/NodeClient.java b/core/src/main/java/org/elasticsearch/client/node/NodeClient.java index 65adfad64d..350b452dc2 100644 --- a/core/src/main/java/org/elasticsearch/client/node/NodeClient.java +++ b/core/src/main/java/org/elasticsearch/client/node/NodeClient.java @@ -27,7 +27,6 @@ import org.elasticsearch.action.ActionResponse; import org.elasticsearch.action.GenericAction; import org.elasticsearch.action.support.TransportAction; import org.elasticsearch.client.support.AbstractClient; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.common.inject.Inject; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.threadpool.ThreadPool; @@ -44,8 +43,8 @@ public class NodeClient extends AbstractClient { private final Map<GenericAction, TransportAction> actions; @Inject - public NodeClient(Settings settings, ThreadPool threadPool, Headers headers, Map<GenericAction, TransportAction> actions) { - super(settings, threadPool, headers); + public NodeClient(Settings settings, ThreadPool threadPool, Map<GenericAction, TransportAction> actions) { + super(settings, threadPool); this.actions = unmodifiableMap(actions); } diff --git a/core/src/main/java/org/elasticsearch/client/node/NodeClientModule.java b/core/src/main/java/org/elasticsearch/client/node/NodeClientModule.java index fb0891da8c..de13488730 100644 --- a/core/src/main/java/org/elasticsearch/client/node/NodeClientModule.java +++ b/core/src/main/java/org/elasticsearch/client/node/NodeClientModule.java @@ -20,7 +20,6 @@ package org.elasticsearch.client.node; import org.elasticsearch.client.Client; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.common.inject.AbstractModule; /** @@ -30,7 +29,6 @@ public class NodeClientModule extends AbstractModule { @Override protected void configure() { - bind(Headers.class).asEagerSingleton(); bind(Client.class).to(NodeClient.class).asEagerSingleton(); } } diff --git a/core/src/main/java/org/elasticsearch/client/support/AbstractClient.java b/core/src/main/java/org/elasticsearch/client/support/AbstractClient.java index e5a465442b..5368102530 100644 --- a/core/src/main/java/org/elasticsearch/client/support/AbstractClient.java +++ b/core/src/main/java/org/elasticsearch/client/support/AbstractClient.java @@ -317,12 +317,16 @@ import org.elasticsearch.client.AdminClient; import org.elasticsearch.client.Client; import org.elasticsearch.client.ClusterAdminClient; import org.elasticsearch.client.ElasticsearchClient; +import org.elasticsearch.client.FilterClient; import org.elasticsearch.client.IndicesAdminClient; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.component.AbstractComponent; import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.threadpool.ThreadPool; +import java.util.Map; + /** * */ @@ -330,24 +334,16 @@ public abstract class AbstractClient extends AbstractComponent implements Client private final ThreadPool threadPool; private final Admin admin; - - private final Headers headers; private final ThreadedActionListener.Wrapper threadedWrapper; - public AbstractClient(Settings settings, ThreadPool threadPool, Headers headers) { + public AbstractClient(Settings settings, ThreadPool threadPool) { super(settings); this.threadPool = threadPool; - this.headers = headers; this.admin = new Admin(this); this.threadedWrapper = new ThreadedActionListener.Wrapper(logger, settings, threadPool); } @Override - public Headers headers() { - return this.headers; - } - - @Override public final Settings settings() { return this.settings; } @@ -379,7 +375,6 @@ public abstract class AbstractClient extends AbstractComponent implements Client */ @Override public final <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void execute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) { - headers.applyTo(request); listener = threadedWrapper.wrap(listener); doExecute(action, request, listener); } @@ -1672,4 +1667,18 @@ public abstract class AbstractClient extends AbstractComponent implements Client execute(GetSettingsAction.INSTANCE, request, listener); } } + + @Override + public Client filterWithHeader(Map<String, String> headers) { + return new FilterClient(this) { + @Override + protected <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void doExecute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) { + ThreadContext threadContext = threadPool().getThreadContext(); + try (ThreadContext.StoredContext ctx = threadContext.stashContext()) { + threadContext.putHeader(headers); + super.doExecute(action, request, listener); + } + } + }; + } } diff --git a/core/src/main/java/org/elasticsearch/client/support/Headers.java b/core/src/main/java/org/elasticsearch/client/support/Headers.java deleted file mode 100644 index f46bd0a1c9..0000000000 --- a/core/src/main/java/org/elasticsearch/client/support/Headers.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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.client.support; - -import org.elasticsearch.common.inject.Inject; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.transport.TransportMessage; - -/** - * Client request headers picked up from the client settings. Applied to every - * request sent by the client (both transport and node clients) - */ -public class Headers { - - public static final String PREFIX = "request.headers"; - - public static final Headers EMPTY = new Headers(Settings.EMPTY) { - @Override - public <M extends TransportMessage<?>> M applyTo(M message) { - return message; - } - }; - - private final Settings headers; - - @Inject - public Headers(Settings settings) { - headers = resolveHeaders(settings); - } - - public <M extends TransportMessage<?>> M applyTo(M message) { - for (String key : headers.names()) { - if (!message.hasHeader(key)) { - message.putHeader(key, headers.get(key)); - } - } - return message; - } - - public Settings headers() { - return headers; - } - - static Settings resolveHeaders(Settings settings) { - Settings headers = settings.getAsSettings(PREFIX); - return headers != null ? headers : Settings.EMPTY; - } -} diff --git a/core/src/main/java/org/elasticsearch/client/transport/TransportClient.java b/core/src/main/java/org/elasticsearch/client/transport/TransportClient.java index 3b8be668f4..1a380c7ab3 100644 --- a/core/src/main/java/org/elasticsearch/client/transport/TransportClient.java +++ b/core/src/main/java/org/elasticsearch/client/transport/TransportClient.java @@ -28,7 +28,6 @@ import org.elasticsearch.action.ActionRequestBuilder; import org.elasticsearch.action.ActionResponse; import org.elasticsearch.cache.recycler.PageCacheRecycler; import org.elasticsearch.client.support.AbstractClient; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.client.transport.support.TransportProxyClient; import org.elasticsearch.cluster.ClusterNameModule; import org.elasticsearch.cluster.node.DiscoveryNode; @@ -173,7 +172,7 @@ public class TransportClient extends AbstractClient { private final TransportProxyClient proxy; private TransportClient(Injector injector) { - super(injector.getInstance(Settings.class), injector.getInstance(ThreadPool.class), injector.getInstance(Headers.class)); + super(injector.getInstance(Settings.class), injector.getInstance(ThreadPool.class)); this.injector = injector; nodesService = injector.getInstance(TransportClientNodesService.class); proxy = injector.getInstance(TransportProxyClient.class); diff --git a/core/src/main/java/org/elasticsearch/client/transport/TransportClientNodesService.java b/core/src/main/java/org/elasticsearch/client/transport/TransportClientNodesService.java index 56befbb9b8..fcbd122ef7 100644 --- a/core/src/main/java/org/elasticsearch/client/transport/TransportClientNodesService.java +++ b/core/src/main/java/org/elasticsearch/client/transport/TransportClientNodesService.java @@ -29,7 +29,6 @@ import org.elasticsearch.action.admin.cluster.node.liveness.TransportLivenessAct import org.elasticsearch.action.admin.cluster.state.ClusterStateAction; import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.client.Requests; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.cluster.ClusterName; import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.component.AbstractComponent; @@ -79,8 +78,6 @@ public class TransportClientNodesService extends AbstractComponent { private final Version minCompatibilityVersion; - private final Headers headers; - // nodes that are added to be discovered private volatile List<DiscoveryNode> listedNodes = Collections.emptyList(); @@ -103,13 +100,12 @@ public class TransportClientNodesService extends AbstractComponent { @Inject public TransportClientNodesService(Settings settings, ClusterName clusterName, TransportService transportService, - ThreadPool threadPool, Headers headers, Version version) { + ThreadPool threadPool, Version version) { super(settings); this.clusterName = clusterName; this.transportService = transportService; this.threadPool = threadPool; this.minCompatibilityVersion = version.minimumCompatibilityVersion(); - this.headers = headers; this.nodesSamplerInterval = this.settings.getAsTime("client.transport.nodes_sampler_interval", timeValueSeconds(5)); this.pingTimeout = this.settings.getAsTime("client.transport.ping_timeout", timeValueSeconds(5)).millis(); @@ -358,7 +354,7 @@ public class TransportClientNodesService extends AbstractComponent { } try { LivenessResponse livenessResponse = transportService.submitRequest(listedNode, TransportLivenessAction.NAME, - headers.applyTo(new LivenessRequest()), + new LivenessRequest(), TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STATE).withTimeout(pingTimeout).build(), new FutureTransportResponseHandler<LivenessResponse>() { @Override @@ -428,8 +424,7 @@ public class TransportClientNodesService extends AbstractComponent { return; } } - transportService.sendRequest(listedNode, ClusterStateAction.NAME, - headers.applyTo(Requests.clusterStateRequest().clear().nodes(true).local(true)), + transportService.sendRequest(listedNode, ClusterStateAction.NAME, Requests.clusterStateRequest().clear().nodes(true).local(true), TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STATE).withTimeout(pingTimeout).build(), new BaseTransportResponseHandler<ClusterStateResponse>() { diff --git a/core/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java b/core/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java index df85762357..dd30a71168 100644 --- a/core/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java +++ b/core/src/main/java/org/elasticsearch/cluster/ClusterStateObserver.java @@ -23,6 +23,7 @@ import org.elasticsearch.ElasticsearchException; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.logging.ESLogger; import org.elasticsearch.common.unit.TimeValue; +import org.elasticsearch.common.util.concurrent.ThreadContext; import java.util.concurrent.atomic.AtomicReference; @@ -44,6 +45,7 @@ public class ClusterStateObserver { }; private final ClusterService clusterService; + private final ThreadContext contextHolder; volatile TimeValue timeOutValue; @@ -55,8 +57,8 @@ public class ClusterStateObserver { volatile boolean timedOut; - public ClusterStateObserver(ClusterService clusterService, ESLogger logger) { - this(clusterService, new TimeValue(60000), logger); + public ClusterStateObserver(ClusterService clusterService, ESLogger logger, ThreadContext contextHolder) { + this(clusterService, new TimeValue(60000), logger, contextHolder); } /** @@ -64,7 +66,7 @@ public class ClusterStateObserver { * will fail any existing or new #waitForNextChange calls. Set to null * to wait indefinitely */ - public ClusterStateObserver(ClusterService clusterService, @Nullable TimeValue timeout, ESLogger logger) { + public ClusterStateObserver(ClusterService clusterService, @Nullable TimeValue timeout, ESLogger logger, ThreadContext contextHolder) { this.clusterService = clusterService; this.lastObservedState = new AtomicReference<>(new ObservedState(clusterService.state())); this.timeOutValue = timeout; @@ -72,6 +74,7 @@ public class ClusterStateObserver { this.startTimeNS = System.nanoTime(); } this.logger = logger; + this.contextHolder = contextHolder; } /** last cluster state observer by this observer. Note that this may not be the current one */ @@ -146,7 +149,7 @@ public class ClusterStateObserver { listener.onNewClusterState(newState.clusterState); } else { logger.trace("observer: sampled state rejected by predicate ({}). adding listener to ClusterService", newState); - ObservingContext context = new ObservingContext(listener, changePredicate); + ObservingContext context = new ObservingContext(new ContextPreservingListener(listener, contextHolder.newStoredContext()), changePredicate); if (!observingContext.compareAndSet(null, context)) { throw new ElasticsearchException("already waiting for a cluster state change"); } @@ -317,4 +320,33 @@ public class ClusterStateObserver { return "version [" + clusterState.version() + "], status [" + status + "]"; } } + + private final static class ContextPreservingListener implements Listener { + private final Listener delegate; + private final ThreadContext.StoredContext tempContext; + + + private ContextPreservingListener(Listener delegate, ThreadContext.StoredContext storedContext) { + this.tempContext = storedContext; + this.delegate = delegate; + } + + @Override + public void onNewClusterState(ClusterState state) { + tempContext.restore(); + delegate.onNewClusterState(state); + } + + @Override + public void onClusterServiceClose() { + tempContext.restore(); + delegate.onClusterServiceClose(); + } + + @Override + public void onTimeout(TimeValue timeout) { + tempContext.restore(); + delegate.onTimeout(timeout); + } + } } diff --git a/core/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java b/core/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java index ca135728b8..c814fb73ed 100644 --- a/core/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java +++ b/core/src/main/java/org/elasticsearch/cluster/service/InternalClusterService.java @@ -190,7 +190,7 @@ public class InternalClusterService extends AbstractLifecycleComponent<ClusterSe protected void doStart() { add(localNodeMasterListeners); this.clusterState = ClusterState.builder(clusterState).blocks(initialBlocks).build(); - this.updateTasksExecutor = EsExecutors.newSinglePrioritizing(UPDATE_THREAD_NAME, daemonThreadFactory(settings, UPDATE_THREAD_NAME)); + this.updateTasksExecutor = EsExecutors.newSinglePrioritizing(UPDATE_THREAD_NAME, daemonThreadFactory(settings, UPDATE_THREAD_NAME), threadPool.getThreadContext()); this.reconnectToNodes = threadPool.schedule(reconnectInterval, ThreadPool.Names.GENERIC, new ReconnectToNodes()); Map<String, String> nodeAttributes = discoveryNodeService.buildAttributes(); // note, we rely on the fact that its a new id each time we start, see FD and "kill -9" handling diff --git a/core/src/main/java/org/elasticsearch/common/ContextAndHeaderHolder.java b/core/src/main/java/org/elasticsearch/common/ContextAndHeaderHolder.java deleted file mode 100644 index 9a3140dba1..0000000000 --- a/core/src/main/java/org/elasticsearch/common/ContextAndHeaderHolder.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * 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.common; - -import com.carrotsearch.hppc.ObjectObjectAssociativeContainer; -import com.carrotsearch.hppc.ObjectObjectHashMap; -import org.elasticsearch.common.collect.ImmutableOpenMap; - -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; -import java.util.Set; - -/** - * - */ -public class ContextAndHeaderHolder implements HasContextAndHeaders { - - private ObjectObjectHashMap<Object, Object> context; - protected Map<String, Object> headers; - - @SuppressWarnings("unchecked") - @Override - public final synchronized <V> V putInContext(Object key, Object value) { - if (context == null) { - context = new ObjectObjectHashMap<>(2); - } - return (V) context.put(key, value); - } - - @Override - public final synchronized void putAllInContext(ObjectObjectAssociativeContainer<Object, Object> map) { - if (map == null) { - return; - } - if (context == null) { - context = new ObjectObjectHashMap<>(map); - } else { - context.putAll(map); - } - } - - @SuppressWarnings("unchecked") - @Override - public final synchronized <V> V getFromContext(Object key) { - return context != null ? (V) context.get(key) : null; - } - - @SuppressWarnings("unchecked") - @Override - public final synchronized <V> V getFromContext(Object key, V defaultValue) { - V value = getFromContext(key); - return value == null ? defaultValue : value; - } - - @Override - public final synchronized boolean hasInContext(Object key) { - return context != null && context.containsKey(key); - } - - @Override - public final synchronized int contextSize() { - return context != null ? context.size() : 0; - } - - @Override - public final synchronized boolean isContextEmpty() { - return context == null || context.isEmpty(); - } - - @Override - public synchronized ImmutableOpenMap<Object, Object> getContext() { - return context != null ? ImmutableOpenMap.copyOf(context) : ImmutableOpenMap.of(); - } - - @Override - public synchronized void copyContextFrom(HasContext other) { - if (other == null) { - return; - } - - synchronized (other) { - ImmutableOpenMap<Object, Object> otherContext = other.getContext(); - if (otherContext == null) { - return; - } - if (context == null) { - ObjectObjectHashMap<Object, Object> map = new ObjectObjectHashMap<>(other.getContext().size()); - map.putAll(otherContext); - this.context = map; - } else { - context.putAll(otherContext); - } - } - } - - @SuppressWarnings("unchecked") - @Override - public final void putHeader(String key, Object value) { - if (headers == null) { - headers = new HashMap<>(); - } - headers.put(key, value); - } - - @SuppressWarnings("unchecked") - @Override - public final <V> V getHeader(String key) { - return headers != null ? (V) headers.get(key) : null; - } - - @Override - public final boolean hasHeader(String key) { - return headers != null && headers.containsKey(key); - } - - @Override - public Set<String> getHeaders() { - return headers != null ? headers.keySet() : Collections.<String>emptySet(); - } - - @Override - public void copyHeadersFrom(HasHeaders from) { - if (from != null && from.getHeaders() != null && !from.getHeaders().isEmpty()) { - for (String headerName : from.getHeaders()) { - putHeader(headerName, from.getHeader(headerName)); - } - } - } - - @Override - public void copyContextAndHeadersFrom(HasContextAndHeaders other) { - copyContextFrom(other); - copyHeadersFrom(other); - } -} diff --git a/core/src/main/java/org/elasticsearch/common/DelegatingHasContextAndHeaders.java b/core/src/main/java/org/elasticsearch/common/DelegatingHasContextAndHeaders.java deleted file mode 100644 index 52d5af5e33..0000000000 --- a/core/src/main/java/org/elasticsearch/common/DelegatingHasContextAndHeaders.java +++ /dev/null @@ -1,111 +0,0 @@ -/* - * 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.common; - -import com.carrotsearch.hppc.ObjectObjectAssociativeContainer; -import org.elasticsearch.common.collect.ImmutableOpenMap; - -import java.util.Set; - -public class DelegatingHasContextAndHeaders implements HasContextAndHeaders { - - private HasContextAndHeaders delegate; - - public DelegatingHasContextAndHeaders(HasContextAndHeaders delegate) { - this.delegate = delegate; - } - - @Override - public <V> void putHeader(String key, V value) { - delegate.putHeader(key, value); - } - - @Override - public void copyContextAndHeadersFrom(HasContextAndHeaders other) { - delegate.copyContextAndHeadersFrom(other); - } - - @Override - public <V> V getHeader(String key) { - return delegate.getHeader(key); - } - - @Override - public boolean hasHeader(String key) { - return delegate.hasHeader(key); - } - - @Override - public <V> V putInContext(Object key, Object value) { - return delegate.putInContext(key, value); - } - - @Override - public Set<String> getHeaders() { - return delegate.getHeaders(); - } - - @Override - public void copyHeadersFrom(HasHeaders from) { - delegate.copyHeadersFrom(from); - } - - @Override - public void putAllInContext(ObjectObjectAssociativeContainer<Object, Object> map) { - delegate.putAllInContext(map); - } - - @Override - public <V> V getFromContext(Object key) { - return delegate.getFromContext(key); - } - - @Override - public <V> V getFromContext(Object key, V defaultValue) { - return delegate.getFromContext(key, defaultValue); - } - - @Override - public boolean hasInContext(Object key) { - return delegate.hasInContext(key); - } - - @Override - public int contextSize() { - return delegate.contextSize(); - } - - @Override - public boolean isContextEmpty() { - return delegate.isContextEmpty(); - } - - @Override - public ImmutableOpenMap<Object, Object> getContext() { - return delegate.getContext(); - } - - @Override - public void copyContextFrom(HasContext other) { - delegate.copyContextFrom(other); - } - - -} diff --git a/core/src/main/java/org/elasticsearch/common/HasContext.java b/core/src/main/java/org/elasticsearch/common/HasContext.java deleted file mode 100644 index 6a303e39ba..0000000000 --- a/core/src/main/java/org/elasticsearch/common/HasContext.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * 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.common; - -import com.carrotsearch.hppc.ObjectObjectAssociativeContainer; -import org.elasticsearch.common.collect.ImmutableOpenMap; - -public interface HasContext { - - /** - * Attaches the given value to the context. - * - * @return The previous value that was associated with the given key in the context, or - * {@code null} if there was none. - */ - <V> V putInContext(Object key, Object value); - - /** - * Attaches the given values to the context - */ - void putAllInContext(ObjectObjectAssociativeContainer<Object, Object> map); - - /** - * @return The context value that is associated with the given key - * - * @see #putInContext(Object, Object) - */ - <V> V getFromContext(Object key); - - /** - * @param defaultValue The default value that should be returned for the given key, if no - * value is currently associated with it. - * - * @return The value that is associated with the given key in the context - * - * @see #putInContext(Object, Object) - */ - <V> V getFromContext(Object key, V defaultValue); - - /** - * Checks if the context contains an entry with the given key - */ - boolean hasInContext(Object key); - - /** - * @return The number of values attached in the context. - */ - int contextSize(); - - /** - * Checks if the context is empty. - */ - boolean isContextEmpty(); - - /** - * @return A safe immutable copy of the current context. - */ - ImmutableOpenMap<Object, Object> getContext(); - - /** - * Copies the context from the given context holder to this context holder. Any shared keys between - * the two context will be overridden by the given context holder. - */ - void copyContextFrom(HasContext other); -} diff --git a/core/src/main/java/org/elasticsearch/common/HasContextAndHeaders.java b/core/src/main/java/org/elasticsearch/common/HasContextAndHeaders.java deleted file mode 100644 index 35bea9a777..0000000000 --- a/core/src/main/java/org/elasticsearch/common/HasContextAndHeaders.java +++ /dev/null @@ -1,33 +0,0 @@ -/* - * 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.common; - -/** - * marker interface - */ -public interface HasContextAndHeaders extends HasContext, HasHeaders { - - /** - * copies over the context and the headers - * @param other another object supporting headers and context - */ - void copyContextAndHeadersFrom(HasContextAndHeaders other); - -} diff --git a/core/src/main/java/org/elasticsearch/common/HasHeaders.java b/core/src/main/java/org/elasticsearch/common/HasHeaders.java deleted file mode 100644 index ab3a7da643..0000000000 --- a/core/src/main/java/org/elasticsearch/common/HasHeaders.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * 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.common; - -import java.util.Set; - -/** - * - */ -public interface HasHeaders { - - <V> void putHeader(String key, V value); - - <V> V getHeader(String key); - - boolean hasHeader(String key); - - Set<String> getHeaders(); - - void copyHeadersFrom(HasHeaders from); -} diff --git a/core/src/main/java/org/elasticsearch/common/network/NetworkModule.java b/core/src/main/java/org/elasticsearch/common/network/NetworkModule.java index b3abed6e23..089e67d8f6 100644 --- a/core/src/main/java/org/elasticsearch/common/network/NetworkModule.java +++ b/core/src/main/java/org/elasticsearch/common/network/NetworkModule.java @@ -19,7 +19,6 @@ package org.elasticsearch.common.network; -import org.elasticsearch.client.support.Headers; import org.elasticsearch.client.transport.TransportClientNodesService; import org.elasticsearch.client.transport.support.TransportProxyClient; import org.elasticsearch.cluster.node.DiscoveryNode; @@ -361,7 +360,6 @@ public class NetworkModule extends AbstractModule { transportTypes.bindType(binder(), settings, TRANSPORT_TYPE_KEY, defaultTransport); if (transportClient) { - bind(Headers.class).asEagerSingleton(); bind(TransportProxyClient.class).asEagerSingleton(); bind(TransportClientNodesService.class).asEagerSingleton(); } else { 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 140f026c35..6111028580 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 @@ -60,30 +60,30 @@ public class EsExecutors { return settings.getAsInt(PROCESSORS, defaultValue); } - public static PrioritizedEsThreadPoolExecutor newSinglePrioritizing(String name, ThreadFactory threadFactory) { - return new PrioritizedEsThreadPoolExecutor(name, 1, 1, 0L, TimeUnit.MILLISECONDS, threadFactory); + public static PrioritizedEsThreadPoolExecutor newSinglePrioritizing(String name, ThreadFactory threadFactory, ThreadContext contextHolder) { + return new PrioritizedEsThreadPoolExecutor(name, 1, 1, 0L, TimeUnit.MILLISECONDS, threadFactory, contextHolder); } - public static EsThreadPoolExecutor newScaling(String name, int min, int max, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) { + public static EsThreadPoolExecutor newScaling(String name, int min, int max, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory, ThreadContext contextHolder) { ExecutorScalingQueue<Runnable> queue = new ExecutorScalingQueue<>(); // we force the execution, since we might run into concurrency issues in offer for ScalingBlockingQueue - EsThreadPoolExecutor executor = new EsThreadPoolExecutor(name, min, max, keepAliveTime, unit, queue, threadFactory, new ForceQueuePolicy()); + EsThreadPoolExecutor executor = new EsThreadPoolExecutor(name, min, max, keepAliveTime, unit, queue, threadFactory, new ForceQueuePolicy(), contextHolder); queue.executor = executor; return executor; } - public static EsThreadPoolExecutor newCached(String name, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) { - return new EsThreadPoolExecutor(name, 0, Integer.MAX_VALUE, keepAliveTime, unit, new SynchronousQueue<Runnable>(), threadFactory, new EsAbortPolicy()); + public static EsThreadPoolExecutor newCached(String name, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory, ThreadContext contextHolder) { + return new EsThreadPoolExecutor(name, 0, Integer.MAX_VALUE, keepAliveTime, unit, new SynchronousQueue<Runnable>(), threadFactory, new EsAbortPolicy(), contextHolder); } - public static EsThreadPoolExecutor newFixed(String name, int size, int queueCapacity, ThreadFactory threadFactory) { + public static EsThreadPoolExecutor newFixed(String name, int size, int queueCapacity, ThreadFactory threadFactory, ThreadContext contextHolder) { BlockingQueue<Runnable> queue; if (queueCapacity < 0) { queue = ConcurrentCollections.newBlockingQueue(); } else { queue = new SizeBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), queueCapacity); } - return new EsThreadPoolExecutor(name, size, size, 0, TimeUnit.MILLISECONDS, queue, threadFactory, new EsAbortPolicy()); + return new EsThreadPoolExecutor(name, size, size, 0, TimeUnit.MILLISECONDS, queue, threadFactory, new EsAbortPolicy(), contextHolder); } public static String threadName(Settings settings, String ... names) { diff --git a/core/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java b/core/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java index 4c02aab1fe..b5417aa238 100644 --- a/core/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java +++ b/core/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java @@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit; */ public class EsThreadPoolExecutor extends ThreadPoolExecutor { + private final ThreadContext contextHolder; private volatile ShutdownListener listener; private final Object monitor = new Object(); @@ -38,13 +39,14 @@ public class EsThreadPoolExecutor extends ThreadPoolExecutor { */ private final String name; - EsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { - this(name, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new EsAbortPolicy()); + EsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, ThreadContext contextHolder) { + this(name, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new EsAbortPolicy(), contextHolder); } - EsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, XRejectedExecutionHandler handler) { + EsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, XRejectedExecutionHandler handler, ThreadContext contextHolder) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler); this.name = name; + this.contextHolder = contextHolder; } public void shutdown(ShutdownListener listener) { @@ -80,7 +82,11 @@ public class EsThreadPoolExecutor extends ThreadPoolExecutor { } @Override - public void execute(Runnable command) { + public void execute(final Runnable command) { + doExecute(wrapRunnable(command)); + } + + protected void doExecute(final Runnable command) { try { super.execute(command); } catch (EsRejectedExecutionException ex) { @@ -116,4 +122,94 @@ public class EsThreadPoolExecutor extends ThreadPoolExecutor { b.append(super.toString()).append(']'); return b.toString(); } + + protected Runnable wrapRunnable(Runnable command) { + final Runnable wrappedCommand; + if (command instanceof AbstractRunnable) { + wrappedCommand = new FilterAbstractRunnable(contextHolder, (AbstractRunnable) command); + } else { + wrappedCommand = new FilterRunnable(contextHolder, command); + } + return wrappedCommand; + } + + protected Runnable unwrap(Runnable runnable) { + if (runnable instanceof FilterAbstractRunnable) { + return ((FilterAbstractRunnable) runnable).in; + } else if (runnable instanceof FilterRunnable) { + return ((FilterRunnable) runnable).in; + } + return runnable; + } + + private static class FilterAbstractRunnable extends AbstractRunnable { + private final ThreadContext contextHolder; + private final AbstractRunnable in; + private final ThreadContext.StoredContext ctx; + + FilterAbstractRunnable(ThreadContext contextHolder, AbstractRunnable in) { + this.contextHolder = contextHolder; + ctx = contextHolder.newStoredContext(); + this.in = in; + } + + @Override + public boolean isForceExecution() { + return in.isForceExecution(); + } + + @Override + public void onAfter() { + in.onAfter(); + } + + @Override + public void onFailure(Throwable t) { + in.onFailure(t); + } + + @Override + public void onRejection(Throwable t) { + in.onRejection(t); + } + + @Override + protected void doRun() throws Exception { + try (ThreadContext.StoredContext ingore = contextHolder.stashContext()){ + ctx.restore(); + in.doRun(); + } + } + + @Override + public String toString() { + return in.toString(); + } + + } + + private static class FilterRunnable implements Runnable { + private final ThreadContext contextHolder; + private final Runnable in; + private final ThreadContext.StoredContext ctx; + + FilterRunnable(ThreadContext contextHolder, Runnable in) { + this.contextHolder = contextHolder; + ctx = contextHolder.newStoredContext(); + this.in = in; + } + + @Override + public void run() { + try (ThreadContext.StoredContext ingore = contextHolder.stashContext()){ + ctx.restore(); + in.run(); + } + } + @Override + public String toString() { + return in.toString(); + } + } + } diff --git a/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java b/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java index d0d2906dee..f55c84e943 100644 --- a/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java +++ b/core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java @@ -47,8 +47,8 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { private AtomicLong insertionOrder = new AtomicLong(); private Queue<Runnable> current = ConcurrentCollections.newQueue(); - PrioritizedEsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) { - super(name, corePoolSize, maximumPoolSize, keepAliveTime, unit, new PriorityBlockingQueue<Runnable>(), threadFactory); + PrioritizedEsThreadPoolExecutor(String name, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory, ThreadContext contextHolder) { + super(name, corePoolSize, maximumPoolSize, keepAliveTime, unit, new PriorityBlockingQueue<>(), threadFactory, contextHolder); } public Pending[] getPending() { @@ -88,10 +88,14 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { for (Runnable runnable : runnables) { if (runnable instanceof TieBreakingPrioritizedRunnable) { TieBreakingPrioritizedRunnable t = (TieBreakingPrioritizedRunnable) runnable; - pending.add(new Pending(t.runnable, t.priority(), t.insertionOrder, executing)); + pending.add(new Pending(unwrap(t.runnable), t.priority(), t.insertionOrder, executing)); } else if (runnable instanceof PrioritizedFutureTask) { PrioritizedFutureTask t = (PrioritizedFutureTask) runnable; - pending.add(new Pending(t.task, t.priority, t.insertionOrder, executing)); + Object task = t.task; + if (t.task instanceof Runnable) { + task = unwrap((Runnable) t.task); + } + pending.add(new Pending(task, t.priority, t.insertionOrder, executing)); } } } @@ -107,12 +111,8 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { } public void execute(Runnable command, final ScheduledExecutorService timer, final TimeValue timeout, final Runnable timeoutCallback) { - if (command instanceof PrioritizedRunnable) { - command = new TieBreakingPrioritizedRunnable((PrioritizedRunnable) command, insertionOrder.incrementAndGet()); - } else if (!(command instanceof PrioritizedFutureTask)) { // it might be a callable wrapper... - command = new TieBreakingPrioritizedRunnable(command, Priority.NORMAL, insertionOrder.incrementAndGet()); - } - super.execute(command); + command = wrapRunnable(command); + doExecute(command); if (timeout.nanos() >= 0) { if (command instanceof TieBreakingPrioritizedRunnable) { ((TieBreakingPrioritizedRunnable) command).scheduleTimeout(timer, timeoutCallback, timeout); @@ -125,21 +125,31 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { } @Override - public void execute(Runnable command) { + protected Runnable wrapRunnable(Runnable command) { if (command instanceof PrioritizedRunnable) { - command = new TieBreakingPrioritizedRunnable((PrioritizedRunnable) command, insertionOrder.incrementAndGet()); - } else if (!(command instanceof PrioritizedFutureTask)) { // it might be a callable wrapper... - command = new TieBreakingPrioritizedRunnable(command, Priority.NORMAL, insertionOrder.incrementAndGet()); + if ((command instanceof TieBreakingPrioritizedRunnable)) { + return command; + } + Priority priority = ((PrioritizedRunnable) command).priority(); + return new TieBreakingPrioritizedRunnable(super.wrapRunnable(command), priority, insertionOrder.incrementAndGet()); + } else if (command instanceof PrioritizedFutureTask) { + return command; + } else { // it might be a callable wrapper... + if (command instanceof TieBreakingPrioritizedRunnable) { + return command; + } + return new TieBreakingPrioritizedRunnable(super.wrapRunnable(command), Priority.NORMAL, insertionOrder.incrementAndGet()); } - super.execute(command); } + @Override protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { if (!(runnable instanceof PrioritizedRunnable)) { runnable = PrioritizedRunnable.wrap(runnable, Priority.NORMAL); } - return new PrioritizedFutureTask<>((PrioritizedRunnable) runnable, value, insertionOrder.incrementAndGet()); + Priority priority = ((PrioritizedRunnable) runnable).priority(); + return new PrioritizedFutureTask<>(runnable, priority, value, insertionOrder.incrementAndGet()); } @Override @@ -147,7 +157,7 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { if (!(callable instanceof PrioritizedCallable)) { callable = PrioritizedCallable.wrap(callable, Priority.NORMAL); } - return new PrioritizedFutureTask<>((PrioritizedCallable<T>) callable, insertionOrder.incrementAndGet()); + return new PrioritizedFutureTask<>((PrioritizedCallable)callable, insertionOrder.incrementAndGet()); } public static class Pending { @@ -173,10 +183,6 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { private ScheduledFuture<?> timeoutFuture; private boolean started = false; - TieBreakingPrioritizedRunnable(PrioritizedRunnable runnable, long insertionOrder) { - this(runnable, runnable.priority(), insertionOrder); - } - TieBreakingPrioritizedRunnable(Runnable runnable, Priority priority, long insertionOrder) { super(priority); this.runnable = runnable; @@ -233,6 +239,7 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { runnable = null; timeoutFuture = null; } + } } @@ -242,10 +249,10 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { final Priority priority; final long insertionOrder; - public PrioritizedFutureTask(PrioritizedRunnable runnable, T value, long insertionOrder) { + public PrioritizedFutureTask(Runnable runnable, Priority priority, T value, long insertionOrder) { super(runnable, value); this.task = runnable; - this.priority = runnable.priority(); + this.priority = priority; this.insertionOrder = insertionOrder; } @@ -265,4 +272,5 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor { return insertionOrder < pft.insertionOrder ? -1 : 1; } } + } diff --git a/core/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java b/core/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java new file mode 100644 index 0000000000..5b52771f03 --- /dev/null +++ b/core/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java @@ -0,0 +1,288 @@ +/* + * 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.common.util.concurrent; + +import org.apache.lucene.util.CloseableThreadLocal; +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.common.settings.Settings; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * A ThreadContext a map of string headers and a transient map of keyed objects that are associated with + * a thread. It allows to store and retrieve header information across method calls, network calls as well as threads spawned from a + * thread that has a {@link ThreadContext} associated with. Threads spawned from a {@link org.elasticsearch.threadpool.ThreadPool} have out of the box + * support for {@link ThreadContext} and all threads spawned will inherit the {@link ThreadContext} from the thread that is forking off. + * Network calls will also preserve the senders heaaders automatically. + */ +public final class ThreadContext implements Closeable, Writeable<ThreadContext.ThreadContextStruct>{ + + public static final String PREFIX = "request.headers"; + private final ThreadContextStruct defaultContext; + private final ContextThreadLocal threadLocal; + + /** + * Creates a new ThreadContext instance + * @param settings the settings to read the default request headers from + */ + public ThreadContext(Settings settings) { + Settings headers = settings.getAsSettings(PREFIX); + if (headers == null) { + this.defaultContext = new ThreadContextStruct(Collections.emptyMap()); + } else { + Map<String, String> defaultHeader = new HashMap<>(); + for (String key : headers.names()) { + defaultHeader.put(key, headers.get(key)); + } + this.defaultContext = new ThreadContextStruct(defaultHeader); + } + threadLocal = new ContextThreadLocal(defaultContext); + } + + @Override + public void close() throws IOException { + threadLocal.close(); + } + + /** + * Removes the current context and resets a default context. The removed context can be + * restored when closing the returned {@link StoredContext} + */ + public StoredContext stashContext() { + final ThreadContextStruct context = threadLocal.get(); + threadLocal.set(null); + return () -> { + threadLocal.set(context); + }; + } + + /** + * Just like {@link #stashContext()} but no default context is set. + */ + public StoredContext newStoredContext() { + final ThreadContextStruct context = threadLocal.get(); + return () -> { + threadLocal.set(context); + }; + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + threadLocal.get().writeTo(out); + } + + @Override + public ThreadContextStruct readFrom(StreamInput in) throws IOException { + return defaultContext.readFrom(in); + } + + /** + * Reads the headers from the stream into the current context + */ + public void readHeaders(StreamInput in) throws IOException { + threadLocal.set(readFrom(in)); + } + + + /** + * Returns the header for the given key or <code>null</code> if not present + */ + public String getHeader(String key) { + return threadLocal.get().headers.get(key); + } + + /** + * Returns all of the current contexts headers + */ + public Map<String, String> getHeaders() { + return threadLocal.get().headers; + } + + /** + * Copies all header key, value pairs into the current context + */ + public void copyHeaders(Iterable<Map.Entry<String, String>> headers) { + threadLocal.set(threadLocal.get().copyHeaders(headers)); + } + + /** + * Puts a header into the context + */ + public void putHeader(String key, String value) { + putHeader(Collections.singletonMap(key, value)); + } + + /** + * Puts all of the given headers into this context + */ + public void putHeader(Map<String, String> header) { + threadLocal.set(threadLocal.get().putHeaders(header)); + } + + /** + * Puts a transient header object into this context + */ + public void putTransient(String key, Object value) { + threadLocal.set(threadLocal.get().putTransient(key, value)); + } + + /** + * Returns a transient header object or <code>null</code> if there is no header for the given key + */ + public <T> T getTransient(String key) { + return (T) threadLocal.get().transientHeaders.get(key); + } + + public interface StoredContext extends AutoCloseable { + @Override + void close(); + + default void restore() { + close(); + } + } + + static final class ThreadContextStruct implements Writeable<ThreadContextStruct> { + private final Map<String,String> headers; + private final Map<String, Object> transientHeaders; + + private ThreadContextStruct(StreamInput in) throws IOException { + int numValues = in.readVInt(); + Map<String, String> headers = numValues == 0 ? Collections.emptyMap() : new HashMap<>(numValues); + for (int i = 0; i < numValues; i++) { + headers.put(in.readString(), in.readString()); + } + this.headers = headers; + this.transientHeaders = Collections.emptyMap(); + } + + private ThreadContextStruct(Map<String, String> headers, Map<String, Object> transientHeaders) { + this.headers = headers; + this.transientHeaders = transientHeaders; + } + + private ThreadContextStruct(Map<String, String> headers) { + this(headers, Collections.emptyMap()); + } + + private ThreadContextStruct putHeaders(Map<String, String> headers) { + if (headers.isEmpty()) { + return this; + } else { + Map<String, String> newHeaders = new HashMap<>(this.headers); + newHeaders.putAll(headers); + return new ThreadContextStruct(newHeaders, transientHeaders); + } + } + + private ThreadContextStruct putTransient(String key, Object value) { + Map<String, Object> newTransient = new HashMap<>(this.transientHeaders); + if (newTransient.putIfAbsent(key, value) != null) { + throw new IllegalArgumentException("value for key [" + key + "] already present"); + } + return new ThreadContextStruct(headers, newTransient); + } + + boolean isEmpty() { + return headers.isEmpty() && transientHeaders.isEmpty(); + } + + + private ThreadContextStruct copyHeaders(Iterable<Map.Entry<String, String>> headers) { + Map<String, String> newHeaders = new HashMap<>(); + for (Map.Entry<String, String> header : headers) { + newHeaders.put(header.getKey(), header.getValue()); + } + return putHeaders(newHeaders); + } + + @Override + public ThreadContextStruct readFrom(StreamInput in) throws IOException { + return new ThreadContextStruct(in); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + int keys = headers.size(); + out.writeVInt(keys); + for (Map.Entry<String, String> entry : headers.entrySet()) { + out.writeString(entry.getKey()); + out.writeString(entry.getValue()); + } + } + + } + + private static class ContextThreadLocal extends CloseableThreadLocal<ThreadContextStruct> { + private final ThreadContextStruct defaultStruct; + private final AtomicBoolean closed = new AtomicBoolean(false); + + private ContextThreadLocal(ThreadContextStruct defaultStruct) { + this.defaultStruct = defaultStruct; + } + + @Override + public void set(ThreadContextStruct object) { + try { + if (object == defaultStruct) { + super.set(null); + } else { + super.set(object); + } + } catch (NullPointerException ex) { + ensureOpen(); + throw ex; + } + } + + @Override + public ThreadContextStruct get() { + try { + ThreadContextStruct threadContextStruct = super.get(); + if (threadContextStruct != null) { + return threadContextStruct; + } + return defaultStruct; + } catch (NullPointerException ex) { + ensureOpen(); + throw ex; + } + } + + private void ensureOpen() { + if (closed.get()) { + throw new IllegalStateException("threadcontext is already closed"); + } + } + + @Override + public void close() { + if (closed.compareAndSet(false, true)) { + super.close(); + } + } + } +} diff --git a/core/src/main/java/org/elasticsearch/discovery/zen/ping/unicast/UnicastZenPing.java b/core/src/main/java/org/elasticsearch/discovery/zen/ping/unicast/UnicastZenPing.java index 99feb4b7f7..347229d8e4 100644 --- a/core/src/main/java/org/elasticsearch/discovery/zen/ping/unicast/UnicastZenPing.java +++ b/core/src/main/java/org/elasticsearch/discovery/zen/ping/unicast/UnicastZenPing.java @@ -170,7 +170,7 @@ public class UnicastZenPing extends AbstractLifecycleComponent<ZenPing> implemen transportService.registerRequestHandler(ACTION_NAME, UnicastPingRequest::new, ThreadPool.Names.SAME, new UnicastPingRequestHandler()); ThreadFactory threadFactory = EsExecutors.daemonThreadFactory(settings, "[unicast_connect]"); - unicastConnectExecutor = EsExecutors.newScaling("unicast_connect", 0, concurrentConnects, 60, TimeUnit.SECONDS, threadFactory); + unicastConnectExecutor = EsExecutors.newScaling("unicast_connect", 0, concurrentConnects, 60, TimeUnit.SECONDS, threadFactory, threadPool.getThreadContext()); } @Override diff --git a/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java b/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java index a117eb709a..fb174f4bd4 100644 --- a/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java +++ b/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayMetaState.java @@ -183,7 +183,7 @@ public class TransportNodesListGatewayMetaState extends TransportNodesAction<Tra } NodeRequest(String nodeId, TransportNodesListGatewayMetaState.Request request) { - super(request, nodeId); + super(nodeId); } @Override diff --git a/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java b/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java index 27ee0c17da..464e08709c 100644 --- a/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java +++ b/core/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java @@ -247,7 +247,7 @@ public class TransportNodesListGatewayStartedShards extends TransportNodesAction } NodeRequest(String nodeId, TransportNodesListGatewayStartedShards.Request request) { - super(request, nodeId); + super(nodeId); this.shardId = request.shardId(); this.indexUUID = request.getIndexUUID(); } diff --git a/core/src/main/java/org/elasticsearch/http/HttpServer.java b/core/src/main/java/org/elasticsearch/http/HttpServer.java index 9971ce7722..ada258f0b6 100644 --- a/core/src/main/java/org/elasticsearch/http/HttpServer.java +++ b/core/src/main/java/org/elasticsearch/http/HttpServer.java @@ -24,6 +24,7 @@ import org.elasticsearch.common.inject.Inject; import org.elasticsearch.common.io.FileSystemUtils; import org.elasticsearch.common.io.Streams; import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.env.Environment; import org.elasticsearch.node.service.NodeService; import org.elasticsearch.rest.BytesRestResponse; @@ -53,7 +54,7 @@ import static org.elasticsearch.rest.RestStatus.OK; /** * */ -public class HttpServer extends AbstractLifecycleComponent<HttpServer> { +public class HttpServer extends AbstractLifecycleComponent<HttpServer> implements HttpServerAdapter { private final Environment environment; @@ -79,23 +80,9 @@ public class HttpServer extends AbstractLifecycleComponent<HttpServer> { nodeService.setHttpServer(this); this.disableSites = this.settings.getAsBoolean("http.disable_sites", false); - - transport.httpServerAdapter(new Dispatcher(this)); + transport.httpServerAdapter(this); } - static class Dispatcher implements HttpServerAdapter { - - private final HttpServer server; - - Dispatcher(HttpServer server) { - this.server = server; - } - - @Override - public void dispatchRequest(HttpRequest request, HttpChannel channel) { - server.internalDispatchRequest(request, channel); - } - } @Override protected void doStart() { @@ -125,7 +112,7 @@ public class HttpServer extends AbstractLifecycleComponent<HttpServer> { return transport.stats(); } - public void internalDispatchRequest(final HttpRequest request, final HttpChannel channel) { + public void dispatchRequest(HttpRequest request, HttpChannel channel, ThreadContext threadContext) { String rawPath = request.rawPath(); if (rawPath.startsWith("/_plugin/")) { RestFilterChain filterChain = restController.filterChain(pluginSiteFilter); @@ -135,7 +122,7 @@ public class HttpServer extends AbstractLifecycleComponent<HttpServer> { handleFavicon(request, channel); return; } - restController.dispatchRequest(request, channel); + restController.dispatchRequest(request, channel, threadContext); } diff --git a/core/src/main/java/org/elasticsearch/http/HttpServerAdapter.java b/core/src/main/java/org/elasticsearch/http/HttpServerAdapter.java index a73456f6b5..c49265cae8 100644 --- a/core/src/main/java/org/elasticsearch/http/HttpServerAdapter.java +++ b/core/src/main/java/org/elasticsearch/http/HttpServerAdapter.java @@ -19,10 +19,12 @@ package org.elasticsearch.http; +import org.elasticsearch.common.util.concurrent.ThreadContext; + /** * */ public interface HttpServerAdapter { - void dispatchRequest(HttpRequest request, HttpChannel channel); + void dispatchRequest(HttpRequest request, HttpChannel channel, ThreadContext context); } diff --git a/core/src/main/java/org/elasticsearch/http/netty/HttpRequestHandler.java b/core/src/main/java/org/elasticsearch/http/netty/HttpRequestHandler.java index 5c05efcd17..71d63d8d1d 100644 --- a/core/src/main/java/org/elasticsearch/http/netty/HttpRequestHandler.java +++ b/core/src/main/java/org/elasticsearch/http/netty/HttpRequestHandler.java @@ -19,6 +19,7 @@ package org.elasticsearch.http.netty; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.http.netty.pipelining.OrderedUpstreamMessageEvent; import org.elasticsearch.rest.support.RestUtils; import org.jboss.netty.channel.ChannelHandler; @@ -41,12 +42,14 @@ public class HttpRequestHandler extends SimpleChannelUpstreamHandler { private final Pattern corsPattern; private final boolean httpPipeliningEnabled; private final boolean detailedErrorsEnabled; + private final ThreadContext threadContext; - public HttpRequestHandler(NettyHttpServerTransport serverTransport, boolean detailedErrorsEnabled) { + public HttpRequestHandler(NettyHttpServerTransport serverTransport, boolean detailedErrorsEnabled, ThreadContext threadContext) { this.serverTransport = serverTransport; this.corsPattern = RestUtils.checkCorsSettingForRegex(serverTransport.settings().get(NettyHttpServerTransport.SETTING_CORS_ALLOW_ORIGIN)); this.httpPipeliningEnabled = serverTransport.pipelining; this.detailedErrorsEnabled = detailedErrorsEnabled; + this.threadContext = threadContext; } @Override @@ -60,6 +63,7 @@ public class HttpRequestHandler extends SimpleChannelUpstreamHandler { request = (HttpRequest) e.getMessage(); } + threadContext.copyHeaders(request.headers()); // the netty HTTP handling always copy over the buffer to its own buffer, either in NioWorker internally // when reading, or using a cumalation buffer NettyHttpRequest httpRequest = new NettyHttpRequest(request, e.getChannel()); diff --git a/core/src/main/java/org/elasticsearch/http/netty/NettyHttpServerTransport.java b/core/src/main/java/org/elasticsearch/http/netty/NettyHttpServerTransport.java index 899bbdc86e..cbab882635 100644 --- a/core/src/main/java/org/elasticsearch/http/netty/NettyHttpServerTransport.java +++ b/core/src/main/java/org/elasticsearch/http/netty/NettyHttpServerTransport.java @@ -38,6 +38,7 @@ import org.elasticsearch.common.unit.ByteSizeUnit; import org.elasticsearch.common.unit.ByteSizeValue; import org.elasticsearch.common.util.BigArrays; import org.elasticsearch.common.util.concurrent.EsExecutors; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.http.BindHttpException; import org.elasticsearch.http.HttpChannel; import org.elasticsearch.http.HttpInfo; @@ -47,6 +48,7 @@ import org.elasticsearch.http.HttpServerTransport; import org.elasticsearch.http.HttpStats; import org.elasticsearch.http.netty.pipelining.HttpPipeliningHandler; import org.elasticsearch.monitor.jvm.JvmInfo; +import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.BindTransportException; import org.jboss.netty.bootstrap.ServerBootstrap; import org.jboss.netty.channel.AdaptiveReceiveBufferSizePredictorFactory; @@ -139,6 +141,7 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer protected final String publishHosts[]; protected final boolean detailedErrorsEnabled; + private final ThreadPool threadPool; protected int publishPort; @@ -167,10 +170,11 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer @Inject @SuppressForbidden(reason = "sets org.jboss.netty.epollBugWorkaround based on netty.epollBugWorkaround") // TODO: why be confusing like this? just let the user do it with the netty parameter instead! - public NettyHttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays) { + public NettyHttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays, ThreadPool threadPool) { super(settings); this.networkService = networkService; this.bigArrays = bigArrays; + this.threadPool = threadPool; if (settings.getAsBoolean("netty.epollBugWorkaround", false)) { System.setProperty("org.jboss.netty.epollBugWorkaround", "true"); @@ -389,7 +393,7 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer } protected void dispatchRequest(HttpRequest request, HttpChannel channel) { - httpServerAdapter.dispatchRequest(request, channel); + httpServerAdapter.dispatchRequest(request, channel, threadPool.getThreadContext()); } protected void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { @@ -414,7 +418,7 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer } public ChannelPipelineFactory configureServerChannelPipelineFactory() { - return new HttpChannelPipelineFactory(this, detailedErrorsEnabled); + return new HttpChannelPipelineFactory(this, detailedErrorsEnabled, threadPool.getThreadContext()); } protected static class HttpChannelPipelineFactory implements ChannelPipelineFactory { @@ -422,9 +426,9 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer protected final NettyHttpServerTransport transport; protected final HttpRequestHandler requestHandler; - public HttpChannelPipelineFactory(NettyHttpServerTransport transport, boolean detailedErrorsEnabled) { + public HttpChannelPipelineFactory(NettyHttpServerTransport transport, boolean detailedErrorsEnabled, ThreadContext threadContext) { this.transport = transport; - this.requestHandler = new HttpRequestHandler(transport, detailedErrorsEnabled); + this.requestHandler = new HttpRequestHandler(transport, detailedErrorsEnabled, threadContext); } @Override diff --git a/core/src/main/java/org/elasticsearch/index/query/GeoShapeQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/GeoShapeQueryBuilder.java index 454465727b..f7d8b22d78 100644 --- a/core/src/main/java/org/elasticsearch/index/query/GeoShapeQueryBuilder.java +++ b/core/src/main/java/org/elasticsearch/index/query/GeoShapeQueryBuilder.java @@ -240,7 +240,6 @@ public class GeoShapeQueryBuilder extends AbstractQueryBuilder<GeoShapeQueryBuil ShapeBuilder shapeToQuery = shape; if (shapeToQuery == null) { GetRequest getRequest = new GetRequest(indexedShapeIndex, indexedShapeType, indexedShapeId); - getRequest.copyContextAndHeadersFrom(SearchContext.current()); shapeToQuery = fetch(context.getClient(), getRequest, indexedShapePath); } MappedFieldType fieldType = context.fieldMapper(fieldName); diff --git a/core/src/main/java/org/elasticsearch/index/query/MoreLikeThisQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/MoreLikeThisQueryBuilder.java index ffb21a32dd..4224ee38c6 100644 --- a/core/src/main/java/org/elasticsearch/index/query/MoreLikeThisQueryBuilder.java +++ b/core/src/main/java/org/elasticsearch/index/query/MoreLikeThisQueryBuilder.java @@ -917,7 +917,6 @@ public class MoreLikeThisQueryBuilder extends AbstractQueryBuilder<MoreLikeThisQ for (Item item : unlikeItems) { request.add(item.toTermVectorsRequest()); } - request.copyContextAndHeadersFrom(searchContext); return client.multiTermVectors(request).actionGet(); } diff --git a/core/src/main/java/org/elasticsearch/index/query/QueryShardContext.java b/core/src/main/java/org/elasticsearch/index/query/QueryShardContext.java index 3c2ab5b0d8..a5c4ebb65c 100644 --- a/core/src/main/java/org/elasticsearch/index/query/QueryShardContext.java +++ b/core/src/main/java/org/elasticsearch/index/query/QueryShardContext.java @@ -364,8 +364,8 @@ public class QueryShardContext { /* * Executes the given template, and returns the response. */ - public BytesReference executeQueryTemplate(Template template, SearchContext searchContext) { - ExecutableScript executable = getScriptService().executable(template, ScriptContext.Standard.SEARCH, searchContext, Collections.emptyMap()); + public BytesReference executeQueryTemplate(Template template) { + ExecutableScript executable = getScriptService().executable(template, ScriptContext.Standard.SEARCH, Collections.emptyMap()); return (BytesReference) executable.run(); } diff --git a/core/src/main/java/org/elasticsearch/index/query/TemplateQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/TemplateQueryBuilder.java index 59ff19748a..02a9bc42e3 100644 --- a/core/src/main/java/org/elasticsearch/index/query/TemplateQueryBuilder.java +++ b/core/src/main/java/org/elasticsearch/index/query/TemplateQueryBuilder.java @@ -100,7 +100,7 @@ public class TemplateQueryBuilder extends AbstractQueryBuilder<TemplateQueryBuil @Override protected Query doToQuery(QueryShardContext context) throws IOException { - BytesReference querySource = context.executeQueryTemplate(template, SearchContext.current()); + BytesReference querySource = context.executeQueryTemplate(template); try (XContentParser qSourceParser = XContentFactory.xContent(querySource).createParser(querySource)) { final QueryShardContext contextCopy = new QueryShardContext(context); contextCopy.reset(qSourceParser); diff --git a/core/src/main/java/org/elasticsearch/index/query/TermsQueryBuilder.java b/core/src/main/java/org/elasticsearch/index/query/TermsQueryBuilder.java index 388a21c7a5..f91c49cfc0 100644 --- a/core/src/main/java/org/elasticsearch/index/query/TermsQueryBuilder.java +++ b/core/src/main/java/org/elasticsearch/index/query/TermsQueryBuilder.java @@ -249,7 +249,6 @@ public class TermsQueryBuilder extends AbstractQueryBuilder<TermsQueryBuilder> { List<Object> terms = new ArrayList<>(); GetRequest getRequest = new GetRequest(termsLookup.index(), termsLookup.type(), termsLookup.id()) .preference("_local").routing(termsLookup.routing()); - getRequest.copyContextAndHeadersFrom(SearchContext.current()); final GetResponse getResponse = client.get(getRequest).actionGet(); if (getResponse.isExists()) { List<Object> extractedValues = XContentMapValues.extractRawValues(termsLookup.path(), getResponse.getSourceAsMap()); diff --git a/core/src/main/java/org/elasticsearch/indices/IndicesService.java b/core/src/main/java/org/elasticsearch/indices/IndicesService.java index c3c0ffe473..084f8424d5 100644 --- a/core/src/main/java/org/elasticsearch/indices/IndicesService.java +++ b/core/src/main/java/org/elasticsearch/indices/IndicesService.java @@ -740,5 +740,4 @@ public class IndicesService extends AbstractLifecycleComponent<IndicesService> i public AnalysisRegistry getAnalysis() { return analysisRegistry; } - } diff --git a/core/src/main/java/org/elasticsearch/indices/recovery/RecoverySettings.java b/core/src/main/java/org/elasticsearch/indices/recovery/RecoverySettings.java index c86309db13..8d610dce05 100644 --- a/core/src/main/java/org/elasticsearch/indices/recovery/RecoverySettings.java +++ b/core/src/main/java/org/elasticsearch/indices/recovery/RecoverySettings.java @@ -83,7 +83,6 @@ public class RecoverySettings extends AbstractComponent { this.internalActionLongTimeout = INDICES_RECOVERY_INTERNAL_LONG_ACTION_TIMEOUT_SETTING.get(settings); this.activityTimeout = INDICES_RECOVERY_ACTIVITY_TIMEOUT_SETTING.get(settings); - this.maxBytesPerSec = INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING.get(settings); if (maxBytesPerSec.bytes() <= 0) { rateLimiter = null; diff --git a/core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java b/core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java index f7e683b8f1..b25f16b3e4 100644 --- a/core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java +++ b/core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java @@ -308,7 +308,7 @@ public class RecoveryTarget extends AbstractComponent implements IndexEventListe @Override public void messageReceived(final RecoveryTranslogOperationsRequest request, final TransportChannel channel) throws Exception { try (RecoveriesCollection.StatusRef statusRef = onGoingRecoveries.getStatusSafe(request.recoveryId(), request.shardId())) { - final ClusterStateObserver observer = new ClusterStateObserver(clusterService, null, logger); + final ClusterStateObserver observer = new ClusterStateObserver(clusterService, null, logger, threadPool.getThreadContext()); final RecoveryStatus recoveryStatus = statusRef.status(); final RecoveryState.Translog translog = recoveryStatus.state().getTranslog(); translog.totalOperations(request.totalTranslogOps()); diff --git a/core/src/main/java/org/elasticsearch/indices/store/IndicesStore.java b/core/src/main/java/org/elasticsearch/indices/store/IndicesStore.java index 4a76d26213..30571f09f2 100644 --- a/core/src/main/java/org/elasticsearch/indices/store/IndicesStore.java +++ b/core/src/main/java/org/elasticsearch/indices/store/IndicesStore.java @@ -74,27 +74,23 @@ public class IndicesStore extends AbstractComponent implements ClusterStateListe private final IndicesService indicesService; private final ClusterService clusterService; private final TransportService transportService; + private final ThreadPool threadPool; private TimeValue deleteShardTimeout; @Inject public IndicesStore(Settings settings, IndicesService indicesService, - ClusterService clusterService, TransportService transportService) { + ClusterService clusterService, TransportService transportService, ThreadPool threadPool) { super(settings); this.indicesService = indicesService; this.clusterService = clusterService; this.transportService = transportService; + this.threadPool = threadPool; transportService.registerRequestHandler(ACTION_SHARD_EXISTS, ShardActiveRequest::new, ThreadPool.Names.SAME, new ShardActiveRequestHandler()); this.deleteShardTimeout = settings.getAsTime(INDICES_STORE_DELETE_SHARD_TIMEOUT, new TimeValue(30, TimeUnit.SECONDS)); clusterService.addLast(this); } - IndicesStore() { - super(Settings.EMPTY); - indicesService = null; - this.clusterService = null; - this.transportService = null; - } @Override public void close() { clusterService.remove(this); @@ -278,6 +274,7 @@ public class IndicesStore extends AbstractComponent implements ClusterStateListe @Override public void messageReceived(final ShardActiveRequest request, final TransportChannel channel) throws Exception { IndexShard indexShard = getShard(request); + // make sure shard is really there before register cluster state observer if (indexShard == null) { channel.sendResponse(new ShardActiveResponse(false, clusterService.localNode())); @@ -288,7 +285,7 @@ public class IndicesStore extends AbstractComponent implements ClusterStateListe // in general, using a cluster state observer here is a workaround for the fact that we cannot listen on shard state changes explicitly. // instead we wait for the cluster state changes because we know any shard state change will trigger or be // triggered by a cluster state change. - ClusterStateObserver observer = new ClusterStateObserver(clusterService, request.timeout, logger); + ClusterStateObserver observer = new ClusterStateObserver(clusterService, request.timeout, logger, threadPool.getThreadContext()); // check if shard is active. if so, all is good boolean shardActive = shardActive(indexShard); if (shardActive) { @@ -348,7 +345,6 @@ public class IndicesStore extends AbstractComponent implements ClusterStateListe logger.trace("shard exists request meant for cluster[{}], but this is cluster[{}], ignoring request", request.clusterName, thisClusterName); return null; } - ShardId shardId = request.shardId; IndexService indexService = indicesService.indexService(shardId.index().getName()); if (indexService != null && indexService.indexUUID().equals(request.indexUUID)) { @@ -356,6 +352,7 @@ public class IndicesStore extends AbstractComponent implements ClusterStateListe } return null; } + } private static class ShardActiveRequest extends TransportRequest { diff --git a/core/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java b/core/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java index 65902b443e..54068f19de 100644 --- a/core/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java +++ b/core/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java @@ -344,7 +344,7 @@ public class TransportNodesListShardStoreMetaData extends TransportNodesAction<T } NodeRequest(String nodeId, TransportNodesListShardStoreMetaData.Request request) { - super(request, nodeId); + super(nodeId); this.shardId = request.shardId; this.unallocated = request.unallocated; } diff --git a/core/src/main/java/org/elasticsearch/percolator/PercolateContext.java b/core/src/main/java/org/elasticsearch/percolator/PercolateContext.java index 7d4e18c3d3..639e1386b5 100644 --- a/core/src/main/java/org/elasticsearch/percolator/PercolateContext.java +++ b/core/src/main/java/org/elasticsearch/percolator/PercolateContext.java @@ -31,9 +31,6 @@ import org.apache.lucene.util.Counter; import org.elasticsearch.action.percolate.PercolateShardRequest; import org.elasticsearch.action.search.SearchType; import org.elasticsearch.cache.recycler.PageCacheRecycler; -import org.elasticsearch.common.HasContext; -import org.elasticsearch.common.HasContextAndHeaders; -import org.elasticsearch.common.HasHeaders; import org.elasticsearch.common.ParseFieldMatcher; import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.common.lease.Releasables; @@ -125,7 +122,7 @@ public class PercolateContext extends SearchContext { public PercolateContext(PercolateShardRequest request, SearchShardTarget searchShardTarget, IndexShard indexShard, IndexService indexService, PageCacheRecycler pageCacheRecycler, BigArrays bigArrays, ScriptService scriptService, Query aliasFilter, ParseFieldMatcher parseFieldMatcher) { - super(parseFieldMatcher, request); + super(parseFieldMatcher); this.indexShard = indexShard; this.indexService = indexService; this.fieldDataService = indexService.fieldData(); @@ -146,7 +143,7 @@ public class PercolateContext extends SearchContext { // for testing: PercolateContext(PercolateShardRequest request, SearchShardTarget searchShardTarget, MapperService mapperService) { - super(null, request); + super(null); this.searchShardTarget = searchShardTarget; this.mapperService = mapperService; this.indexService = null; @@ -680,82 +677,6 @@ public class PercolateContext extends SearchContext { } @Override - public <V> V putInContext(Object key, Object value) { - assert false : "percolatecontext does not support contexts & headers"; - return null; - } - - @Override - public void putAllInContext(ObjectObjectAssociativeContainer<Object, Object> map) { - assert false : "percolatocontext does not support contexts & headers"; - } - - @Override - public <V> V getFromContext(Object key) { - return null; - } - - @Override - public <V> V getFromContext(Object key, V defaultValue) { - return defaultValue; - } - - @Override - public boolean hasInContext(Object key) { - return false; - } - - @Override - public int contextSize() { - return 0; - } - - @Override - public boolean isContextEmpty() { - return true; - } - - @Override - public ImmutableOpenMap<Object, Object> getContext() { - return ImmutableOpenMap.of(); - } - - @Override - public void copyContextFrom(HasContext other) { - assert false : "percolatecontext does not support contexts & headers"; - } - - @Override - public <V> void putHeader(String key, V value) { - assert false : "percolatecontext does not support contexts & headers"; - } - - @Override - public <V> V getHeader(String key) { - return null; - } - - @Override - public boolean hasHeader(String key) { - return false; - } - - @Override - public Set<String> getHeaders() { - return Collections.emptySet(); - } - - @Override - public void copyHeadersFrom(HasHeaders from) { - assert false : "percolatecontext does not support contexts & headers"; - } - - @Override - public void copyContextAndHeadersFrom(HasContextAndHeaders other) { - assert false : "percolatecontext does not support contexts & headers"; - } - - @Override public Map<Class<?>, Collector> queryCollectors() { return queryCollectors; } diff --git a/core/src/main/java/org/elasticsearch/percolator/PercolatorService.java b/core/src/main/java/org/elasticsearch/percolator/PercolatorService.java index e6ffa313e8..cb5686ab45 100644 --- a/core/src/main/java/org/elasticsearch/percolator/PercolatorService.java +++ b/core/src/main/java/org/elasticsearch/percolator/PercolatorService.java @@ -39,7 +39,6 @@ import org.elasticsearch.cache.recycler.PageCacheRecycler; import org.elasticsearch.cluster.ClusterService; import org.elasticsearch.cluster.metadata.IndexMetaData; import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.ParseFieldMatcher; import org.elasticsearch.common.component.AbstractComponent; @@ -135,14 +134,14 @@ public class PercolatorService extends AbstractComponent { multi = new MultiDocumentPercolatorIndex(cache); } - public ReduceResult reduce(boolean onlyCount, List<PercolateShardResponse> shardResponses, HasContextAndHeaders headersContext) throws IOException { + public ReduceResult reduce(boolean onlyCount, List<PercolateShardResponse> shardResponses) throws IOException { if (onlyCount) { long finalCount = 0; for (PercolateShardResponse shardResponse : shardResponses) { finalCount += shardResponse.topDocs().totalHits; } - InternalAggregations reducedAggregations = reduceAggregations(shardResponses, headersContext); + InternalAggregations reducedAggregations = reduceAggregations(shardResponses); return new PercolatorService.ReduceResult(finalCount, reducedAggregations); } else { int requestedSize = shardResponses.get(0).requestedSize(); @@ -162,7 +161,7 @@ public class PercolatorService extends AbstractComponent { Map<String, HighlightField> hl = shardResponse.hls().get(doc.doc); matches[i] = new PercolateResponse.Match(new Text(shardResponse.getIndex()), new Text(id), doc.score, hl); } - InternalAggregations reducedAggregations = reduceAggregations(shardResponses, headersContext); + InternalAggregations reducedAggregations = reduceAggregations(shardResponses); return new PercolatorService.ReduceResult(foundMatches, matches, reducedAggregations); } } @@ -307,7 +306,7 @@ public class PercolatorService extends AbstractComponent { cache.close(); } - private InternalAggregations reduceAggregations(List<PercolateShardResponse> shardResults, HasContextAndHeaders headersContext) { + private InternalAggregations reduceAggregations(List<PercolateShardResponse> shardResults) { if (shardResults.get(0).aggregations() == null) { return null; } @@ -316,7 +315,7 @@ public class PercolatorService extends AbstractComponent { for (PercolateShardResponse shardResult : shardResults) { aggregationsList.add(shardResult.aggregations()); } - InternalAggregations aggregations = InternalAggregations.reduce(aggregationsList, new InternalAggregation.ReduceContext(bigArrays, scriptService, headersContext)); + InternalAggregations aggregations = InternalAggregations.reduce(aggregationsList, new InternalAggregation.ReduceContext(bigArrays, scriptService)); if (aggregations != null) { List<SiblingPipelineAggregator> pipelineAggregators = shardResults.get(0).pipelineAggregators(); if (pipelineAggregators != null) { @@ -324,7 +323,7 @@ public class PercolatorService extends AbstractComponent { return (InternalAggregation) p; }).collect(Collectors.toList()); for (SiblingPipelineAggregator pipelineAggregator : pipelineAggregators) { - InternalAggregation newAgg = pipelineAggregator.doReduce(new InternalAggregations(newAggs), new InternalAggregation.ReduceContext(bigArrays, scriptService, headersContext)); + InternalAggregation newAgg = pipelineAggregator.doReduce(new InternalAggregations(newAggs), new InternalAggregation.ReduceContext(bigArrays, scriptService)); newAggs.add(newAgg); } aggregations = new InternalAggregations(newAggs); diff --git a/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java b/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java index 294338c050..befa5c38b1 100644 --- a/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java +++ b/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java @@ -19,19 +19,11 @@ package org.elasticsearch.rest; -import org.elasticsearch.action.Action; -import org.elasticsearch.action.ActionListener; -import org.elasticsearch.action.ActionRequest; -import org.elasticsearch.action.ActionRequestBuilder; -import org.elasticsearch.action.ActionResponse; import org.elasticsearch.client.Client; -import org.elasticsearch.client.FilterClient; import org.elasticsearch.common.ParseFieldMatcher; import org.elasticsearch.common.component.AbstractComponent; import org.elasticsearch.common.settings.Settings; -import java.util.Set; - /** * Base handler for REST requests. * <p> @@ -42,49 +34,19 @@ import java.util.Set; */ public abstract class BaseRestHandler extends AbstractComponent implements RestHandler { - private final RestController controller; private final Client client; protected final ParseFieldMatcher parseFieldMatcher; - protected BaseRestHandler(Settings settings, RestController controller, Client client) { + protected BaseRestHandler(Settings settings, Client client) { super(settings); - this.controller = controller; this.client = client; this.parseFieldMatcher = new ParseFieldMatcher(settings); } @Override public final void handleRequest(RestRequest request, RestChannel channel) throws Exception { - handleRequest(request, channel, new HeadersAndContextCopyClient(client, request, controller.relevantHeaders())); + handleRequest(request, channel, client); } protected abstract void handleRequest(RestRequest request, RestChannel channel, Client client) throws Exception; - - static final class HeadersAndContextCopyClient extends FilterClient { - - private final RestRequest restRequest; - private final Set<String> headers; - - HeadersAndContextCopyClient(Client in, RestRequest restRequest, Set<String> headers) { - super(in); - this.restRequest = restRequest; - this.headers = headers; - } - - private static void copyHeadersAndContext(ActionRequest actionRequest, RestRequest restRequest, Set<String> headers) { - for (String usefulHeader : headers) { - String headerValue = restRequest.header(usefulHeader); - if (headerValue != null) { - actionRequest.putHeader(usefulHeader, headerValue); - } - } - actionRequest.copyContextFrom(restRequest); - } - - @Override - protected <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void doExecute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) { - copyHeadersAndContext(request, restRequest, headers); - super.doExecute(action, request, listener); - } - } } diff --git a/core/src/main/java/org/elasticsearch/rest/RestController.java b/core/src/main/java/org/elasticsearch/rest/RestController.java index d0a46d29f6..64e21002d8 100644 --- a/core/src/main/java/org/elasticsearch/rest/RestController.java +++ b/core/src/main/java/org/elasticsearch/rest/RestController.java @@ -24,13 +24,13 @@ import org.elasticsearch.common.component.AbstractLifecycleComponent; import org.elasticsearch.common.inject.Inject; import org.elasticsearch.common.path.PathTrie; import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.rest.support.RestUtils; import java.io.IOException; import java.util.Arrays; import java.util.Collections; -import java.util.Comparator; import java.util.HashSet; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; @@ -107,12 +107,7 @@ public class RestController extends AbstractLifecycleComponent<RestController> { RestFilter[] copy = new RestFilter[filters.length + 1]; System.arraycopy(filters, 0, copy, 0, filters.length); copy[filters.length] = preProcessor; - Arrays.sort(copy, new Comparator<RestFilter>() { - @Override - public int compare(RestFilter o1, RestFilter o2) { - return Integer.compare(o1.order(), o2.order()); - } - }); + Arrays.sort(copy, (o1, o2) -> Integer.compare(o1.order(), o2.order())); filters = copy; } @@ -163,24 +158,31 @@ public class RestController extends AbstractLifecycleComponent<RestController> { return new ControllerFilterChain(executionFilter); } - public void dispatchRequest(final RestRequest request, final RestChannel channel) { + public void dispatchRequest(final RestRequest request, final RestChannel channel, ThreadContext threadContext) { if (!checkRequestParameters(request, channel)) { return; } - - if (filters.length == 0) { - try { - executeHandler(request, channel); - } catch (Throwable e) { + try (ThreadContext.StoredContext t = threadContext.stashContext()){ + for (String key : relevantHeaders) { + String httpHeader = request.header(key); + if (httpHeader != null) { + threadContext.putHeader(key, httpHeader); + } + } + if (filters.length == 0) { try { - channel.sendResponse(new BytesRestResponse(channel, e)); - } catch (Throwable e1) { - logger.error("failed to send failure response for uri [" + request.uri() + "]", e1); + executeHandler(request, channel); + } catch (Throwable e) { + try { + channel.sendResponse(new BytesRestResponse(channel, e)); + } catch (Throwable e1) { + logger.error("failed to send failure response for uri [" + request.uri() + "]", e1); + } } + } else { + ControllerFilterChain filterChain = new ControllerFilterChain(handlerFilter); + filterChain.continueProcessing(request, channel); } - } else { - ControllerFilterChain filterChain = new ControllerFilterChain(handlerFilter); - filterChain.continueProcessing(request, channel); } } diff --git a/core/src/main/java/org/elasticsearch/rest/RestRequest.java b/core/src/main/java/org/elasticsearch/rest/RestRequest.java index 81f6052db5..8872484d58 100644 --- a/core/src/main/java/org/elasticsearch/rest/RestRequest.java +++ b/core/src/main/java/org/elasticsearch/rest/RestRequest.java @@ -20,7 +20,6 @@ package org.elasticsearch.rest; import org.elasticsearch.common.Booleans; -import org.elasticsearch.common.ContextAndHeaderHolder; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.Strings; import org.elasticsearch.common.bytes.BytesReference; @@ -38,7 +37,7 @@ import static org.elasticsearch.common.unit.TimeValue.parseTimeValue; /** * */ -public abstract class RestRequest extends ContextAndHeaderHolder implements ToXContent.Params { +public abstract class RestRequest implements ToXContent.Params { public enum Method { GET, POST, PUT, DELETE, OPTIONS, HEAD diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/health/RestClusterHealthAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/health/RestClusterHealthAction.java index badf6f6de5..ccd0f98259 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/health/RestClusterHealthAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/health/RestClusterHealthAction.java @@ -43,7 +43,7 @@ public class RestClusterHealthAction extends BaseRestHandler { @Inject public RestClusterHealthAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.GET, "/_cluster/health", this); controller.registerHandler(RestRequest.Method.GET, "/_cluster/health/{index}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/hotthreads/RestNodesHotThreadsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/hotthreads/RestNodesHotThreadsAction.java index 24c4c44941..53bec14f96 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/hotthreads/RestNodesHotThreadsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/hotthreads/RestNodesHotThreadsAction.java @@ -43,7 +43,7 @@ public class RestNodesHotThreadsAction extends BaseRestHandler { @Inject public RestNodesHotThreadsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.GET, "/_cluster/nodes/hotthreads", this); controller.registerHandler(RestRequest.Method.GET, "/_cluster/nodes/hot_threads", this); controller.registerHandler(RestRequest.Method.GET, "/_cluster/nodes/{nodeId}/hotthreads", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/info/RestNodesInfoAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/info/RestNodesInfoAction.java index f2c5185000..ce1e7811da 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/info/RestNodesInfoAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/info/RestNodesInfoAction.java @@ -52,7 +52,7 @@ public class RestNodesInfoAction extends BaseRestHandler { @Inject public RestNodesInfoAction(Settings settings, RestController controller, Client client, SettingsFilter settingsFilter) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_nodes", this); // this endpoint is used for metrics, not for nodeIds, like /_nodes/fs controller.registerHandler(GET, "/_nodes/{nodeId}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/stats/RestNodesStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/stats/RestNodesStatsAction.java index 786891d330..2b3f0518c3 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/stats/RestNodesStatsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/stats/RestNodesStatsAction.java @@ -45,7 +45,7 @@ public class RestNodesStatsAction extends BaseRestHandler { @Inject public RestNodesStatsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_nodes/stats", this); controller.registerHandler(GET, "/_nodes/{nodeId}/stats", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java index 813c782242..46fef04b85 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/tasks/RestListTasksAction.java @@ -37,7 +37,7 @@ public class RestListTasksAction extends BaseRestHandler { @Inject public RestListTasksAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_tasks", this); controller.registerHandler(GET, "/_tasks/{nodeId}", this); controller.registerHandler(GET, "/_tasks/{nodeId}/{actions}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/delete/RestDeleteRepositoryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/delete/RestDeleteRepositoryAction.java index 36e02ba459..136c1cfae3 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/delete/RestDeleteRepositoryAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/delete/RestDeleteRepositoryAction.java @@ -40,7 +40,7 @@ public class RestDeleteRepositoryAction extends BaseRestHandler { @Inject public RestDeleteRepositoryAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(DELETE, "/_snapshot/{repository}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/get/RestGetRepositoriesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/get/RestGetRepositoriesAction.java index fd347ccd33..09422481cf 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/get/RestGetRepositoriesAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/get/RestGetRepositoriesAction.java @@ -50,7 +50,7 @@ public class RestGetRepositoriesAction extends BaseRestHandler { @Inject public RestGetRepositoriesAction(Settings settings, RestController controller, Client client, SettingsFilter settingsFilter) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_snapshot", this); controller.registerHandler(GET, "/_snapshot/{repository}", this); this.settingsFilter = settingsFilter; diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/put/RestPutRepositoryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/put/RestPutRepositoryAction.java index feeeeb77ab..878eb2915b 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/put/RestPutRepositoryAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/put/RestPutRepositoryAction.java @@ -41,7 +41,7 @@ public class RestPutRepositoryAction extends BaseRestHandler { @Inject public RestPutRepositoryAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(PUT, "/_snapshot/{repository}", this); controller.registerHandler(POST, "/_snapshot/{repository}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/verify/RestVerifyRepositoryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/verify/RestVerifyRepositoryAction.java index c0c7ad5b95..306dcbb21b 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/verify/RestVerifyRepositoryAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/repositories/verify/RestVerifyRepositoryAction.java @@ -36,7 +36,7 @@ public class RestVerifyRepositoryAction extends BaseRestHandler { @Inject public RestVerifyRepositoryAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_snapshot/{repository}/_verify", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/reroute/RestClusterRerouteAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/reroute/RestClusterRerouteAction.java index 387728918a..529d73d3e5 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/reroute/RestClusterRerouteAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/reroute/RestClusterRerouteAction.java @@ -49,7 +49,7 @@ public class RestClusterRerouteAction extends BaseRestHandler { @Inject public RestClusterRerouteAction(Settings settings, RestController controller, Client client, SettingsFilter settingsFilter) { - super(settings, controller, client); + super(settings, client); this.settingsFilter = settingsFilter; controller.registerHandler(RestRequest.Method.POST, "/_cluster/reroute", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterGetSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterGetSettingsAction.java index fc4432a658..04bda2cb1f 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterGetSettingsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterGetSettingsAction.java @@ -48,7 +48,7 @@ public class RestClusterGetSettingsAction extends BaseRestHandler { @Inject public RestClusterGetSettingsAction(Settings settings, RestController controller, Client client, ClusterSettings clusterSettings) { - super(settings, controller, client); + super(settings, client); this.clusterSettings = clusterSettings; controller.registerHandler(RestRequest.Method.GET, "/_cluster/settings", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterUpdateSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterUpdateSettingsAction.java index aa84606b07..64083f1e80 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterUpdateSettingsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/settings/RestClusterUpdateSettingsAction.java @@ -43,7 +43,7 @@ public class RestClusterUpdateSettingsAction extends BaseRestHandler { @Inject public RestClusterUpdateSettingsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.PUT, "/_cluster/settings", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/shards/RestClusterSearchShardsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/shards/RestClusterSearchShardsAction.java index ee68c1bbb7..860e110b2d 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/shards/RestClusterSearchShardsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/shards/RestClusterSearchShardsAction.java @@ -42,7 +42,7 @@ public class RestClusterSearchShardsAction extends BaseRestHandler { @Inject public RestClusterSearchShardsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_search_shards", this); controller.registerHandler(POST, "/_search_shards", this); controller.registerHandler(GET, "/{index}/_search_shards", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/create/RestCreateSnapshotAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/create/RestCreateSnapshotAction.java index bf9dd4a011..9d6be664d4 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/create/RestCreateSnapshotAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/create/RestCreateSnapshotAction.java @@ -41,7 +41,7 @@ public class RestCreateSnapshotAction extends BaseRestHandler { @Inject public RestCreateSnapshotAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(PUT, "/_snapshot/{repository}/{snapshot}", this); controller.registerHandler(POST, "/_snapshot/{repository}/{snapshot}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/delete/RestDeleteSnapshotAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/delete/RestDeleteSnapshotAction.java index 66b5a4188c..38c78bd5d8 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/delete/RestDeleteSnapshotAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/delete/RestDeleteSnapshotAction.java @@ -40,7 +40,7 @@ public class RestDeleteSnapshotAction extends BaseRestHandler { @Inject public RestDeleteSnapshotAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(DELETE, "/_snapshot/{repository}/{snapshot}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/get/RestGetSnapshotsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/get/RestGetSnapshotsAction.java index 123798cf99..1151fed8f2 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/get/RestGetSnapshotsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/get/RestGetSnapshotsAction.java @@ -41,7 +41,7 @@ public class RestGetSnapshotsAction extends BaseRestHandler { @Inject public RestGetSnapshotsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_snapshot/{repository}/{snapshot}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/restore/RestRestoreSnapshotAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/restore/RestRestoreSnapshotAction.java index 028285d306..e2a16bd4b4 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/restore/RestRestoreSnapshotAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/restore/RestRestoreSnapshotAction.java @@ -40,7 +40,7 @@ public class RestRestoreSnapshotAction extends BaseRestHandler { @Inject public RestRestoreSnapshotAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_snapshot/{repository}/{snapshot}/_restore", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/status/RestSnapshotsStatusAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/status/RestSnapshotsStatusAction.java index b60a740a15..2e8810e2ba 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/status/RestSnapshotsStatusAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/snapshots/status/RestSnapshotsStatusAction.java @@ -41,7 +41,7 @@ public class RestSnapshotsStatusAction extends BaseRestHandler { @Inject public RestSnapshotsStatusAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_snapshot/{repository}/{snapshot}/_status", this); controller.registerHandler(GET, "/_snapshot/{repository}/_status", this); controller.registerHandler(GET, "/_snapshot/_status", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/state/RestClusterStateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/state/RestClusterStateAction.java index f28ecfe488..720d19a7fe 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/state/RestClusterStateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/state/RestClusterStateAction.java @@ -52,7 +52,7 @@ public class RestClusterStateAction extends BaseRestHandler { @Inject public RestClusterStateAction(Settings settings, RestController controller, Client client, SettingsFilter settingsFilter) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.GET, "/_cluster/state", this); controller.registerHandler(RestRequest.Method.GET, "/_cluster/state/{metric}", this); controller.registerHandler(RestRequest.Method.GET, "/_cluster/state/{metric}/{indices}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/stats/RestClusterStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/stats/RestClusterStatsAction.java index b14293ba31..a09820e71b 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/stats/RestClusterStatsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/stats/RestClusterStatsAction.java @@ -38,7 +38,7 @@ public class RestClusterStatsAction extends BaseRestHandler { @Inject public RestClusterStatsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.GET, "/_cluster/stats", this); controller.registerHandler(RestRequest.Method.GET, "/_cluster/stats/nodes/{nodeId}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/tasks/RestPendingClusterTasksAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/tasks/RestPendingClusterTasksAction.java index 5d9eac430b..333b6d6449 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/tasks/RestPendingClusterTasksAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/tasks/RestPendingClusterTasksAction.java @@ -36,7 +36,7 @@ public class RestPendingClusterTasksAction extends BaseRestHandler { @Inject public RestPendingClusterTasksAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.GET, "/_cluster/pending_tasks", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/RestIndicesAliasesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/RestIndicesAliasesAction.java index f62d6febee..c60671f864 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/RestIndicesAliasesAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/RestIndicesAliasesAction.java @@ -47,7 +47,7 @@ public class RestIndicesAliasesAction extends BaseRestHandler { @Inject public RestIndicesAliasesAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_aliases", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/delete/RestIndexDeleteAliasesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/delete/RestIndexDeleteAliasesAction.java index 6748cc2509..7fcaadc3d8 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/delete/RestIndexDeleteAliasesAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/delete/RestIndexDeleteAliasesAction.java @@ -38,7 +38,7 @@ public class RestIndexDeleteAliasesAction extends BaseRestHandler { @Inject public RestIndexDeleteAliasesAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(DELETE, "/{index}/_alias/{name}", this); controller.registerHandler(DELETE, "/{index}/_aliases/{name}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetAliasesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetAliasesAction.java index aa62ee471d..da439c63d5 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetAliasesAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetAliasesAction.java @@ -52,7 +52,7 @@ public class RestGetAliasesAction extends BaseRestHandler { @Inject public RestGetAliasesAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_alias/{name}", this); controller.registerHandler(GET, "/{index}/_alias/{name}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetIndicesAliasesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetIndicesAliasesAction.java index 4c774b5864..4f9e2b93c2 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetIndicesAliasesAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/get/RestGetIndicesAliasesAction.java @@ -51,7 +51,7 @@ public class RestGetIndicesAliasesAction extends BaseRestHandler { @Inject public RestGetIndicesAliasesAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}/_aliases/{name}", this); controller.registerHandler(GET, "/_aliases/{name}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/head/RestAliasesExistAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/head/RestAliasesExistAction.java index fce40123b6..15ea664245 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/head/RestAliasesExistAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/head/RestAliasesExistAction.java @@ -44,7 +44,7 @@ public class RestAliasesExistAction extends BaseRestHandler { @Inject public RestAliasesExistAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(HEAD, "/_alias/{name}", this); controller.registerHandler(HEAD, "/{index}/_alias/{name}", this); controller.registerHandler(HEAD, "/{index}/_alias", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/put/RestIndexPutAliasAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/put/RestIndexPutAliasAction.java index 4965f6b218..7a0c2ad466 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/put/RestIndexPutAliasAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/alias/put/RestIndexPutAliasAction.java @@ -45,7 +45,7 @@ public class RestIndexPutAliasAction extends BaseRestHandler { @Inject public RestIndexPutAliasAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(PUT, "/{index}/_alias/{name}", this); controller.registerHandler(PUT, "/_alias/{name}", this); controller.registerHandler(PUT, "/{index}/_aliases/{name}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/analyze/RestAnalyzeAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/analyze/RestAnalyzeAction.java index 3a86911f46..e440e1b95c 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/analyze/RestAnalyzeAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/analyze/RestAnalyzeAction.java @@ -61,7 +61,7 @@ public class RestAnalyzeAction extends BaseRestHandler { @Inject public RestAnalyzeAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_analyze", this); controller.registerHandler(GET, "/{index}/_analyze", this); controller.registerHandler(POST, "/_analyze", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/cache/clear/RestClearIndicesCacheAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/cache/clear/RestClearIndicesCacheAction.java index cc06a14b8d..7adb690953 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/cache/clear/RestClearIndicesCacheAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/cache/clear/RestClearIndicesCacheAction.java @@ -51,7 +51,7 @@ public class RestClearIndicesCacheAction extends BaseRestHandler { @Inject public RestClearIndicesCacheAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_cache/clear", this); controller.registerHandler(POST, "/{index}/_cache/clear", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/close/RestCloseIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/close/RestCloseIndexAction.java index 091fbc1680..5f211b88d1 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/close/RestCloseIndexAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/close/RestCloseIndexAction.java @@ -39,7 +39,7 @@ public class RestCloseIndexAction extends BaseRestHandler { @Inject public RestCloseIndexAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.POST, "/_close", this); controller.registerHandler(RestRequest.Method.POST, "/{index}/_close", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/create/RestCreateIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/create/RestCreateIndexAction.java index 41a272cc8b..46bc938897 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/create/RestCreateIndexAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/create/RestCreateIndexAction.java @@ -37,7 +37,7 @@ public class RestCreateIndexAction extends BaseRestHandler { @Inject public RestCreateIndexAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.PUT, "/{index}", this); controller.registerHandler(RestRequest.Method.POST, "/{index}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/delete/RestDeleteIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/delete/RestDeleteIndexAction.java index 0851fb867b..4953842c54 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/delete/RestDeleteIndexAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/delete/RestDeleteIndexAction.java @@ -39,7 +39,7 @@ public class RestDeleteIndexAction extends BaseRestHandler { @Inject public RestDeleteIndexAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.DELETE, "/", this); controller.registerHandler(RestRequest.Method.DELETE, "/{index}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/indices/RestIndicesExistsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/indices/RestIndicesExistsAction.java index 6843f5c5ce..72dea18abd 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/indices/RestIndicesExistsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/indices/RestIndicesExistsAction.java @@ -45,7 +45,7 @@ public class RestIndicesExistsAction extends BaseRestHandler { @Inject public RestIndicesExistsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(HEAD, "/{index}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/types/RestTypesExistsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/types/RestTypesExistsAction.java index f1f227edfd..dd206dcb63 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/types/RestTypesExistsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/exists/types/RestTypesExistsAction.java @@ -44,7 +44,7 @@ public class RestTypesExistsAction extends BaseRestHandler { @Inject public RestTypesExistsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(HEAD, "/{index}/{type}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestFlushAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestFlushAction.java index 47c0451adf..f3b3304bcf 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestFlushAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestFlushAction.java @@ -47,7 +47,7 @@ public class RestFlushAction extends BaseRestHandler { @Inject public RestFlushAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_flush", this); controller.registerHandler(POST, "/{index}/_flush", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestSyncedFlushAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestSyncedFlushAction.java index 4fe893bd41..9bb36f03d6 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestSyncedFlushAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/flush/RestSyncedFlushAction.java @@ -45,7 +45,7 @@ public class RestSyncedFlushAction extends BaseRestHandler { @Inject public RestSyncedFlushAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_flush/synced", this); controller.registerHandler(POST, "/{index}/_flush/synced", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/forcemerge/RestForceMergeAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/forcemerge/RestForceMergeAction.java index d8ef7bace3..8aa2683be5 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/forcemerge/RestForceMergeAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/forcemerge/RestForceMergeAction.java @@ -46,7 +46,7 @@ public class RestForceMergeAction extends BaseRestHandler { @Inject public RestForceMergeAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_forcemerge", this); controller.registerHandler(POST, "/{index}/_forcemerge", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/get/RestGetIndicesAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/get/RestGetIndicesAction.java index b7371f7b80..f32f69b57e 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/get/RestGetIndicesAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/get/RestGetIndicesAction.java @@ -54,7 +54,7 @@ public class RestGetIndicesAction extends BaseRestHandler { @Inject public RestGetIndicesAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}", this); controller.registerHandler(GET, "/{index}/{type}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetFieldMappingAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetFieldMappingAction.java index 7594a097c9..0db931d0a7 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetFieldMappingAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetFieldMappingAction.java @@ -51,7 +51,7 @@ public class RestGetFieldMappingAction extends BaseRestHandler { @Inject public RestGetFieldMappingAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_mapping/field/{fields}", this); controller.registerHandler(GET, "/_mapping/{type}/field/{fields}", this); controller.registerHandler(GET, "/{index}/_mapping/field/{fields}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetMappingAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetMappingAction.java index 48fa60cb4b..09be44648f 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetMappingAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/get/RestGetMappingAction.java @@ -52,7 +52,7 @@ public class RestGetMappingAction extends BaseRestHandler { @Inject public RestGetMappingAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}/{type}/_mapping", this); controller.registerHandler(GET, "/{index}/_mappings/{type}", this); controller.registerHandler(GET, "/{index}/_mapping/{type}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/put/RestPutMappingAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/put/RestPutMappingAction.java index 3ceecbfd3a..fdb16d2fb8 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/put/RestPutMappingAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/mapping/put/RestPutMappingAction.java @@ -44,7 +44,7 @@ public class RestPutMappingAction extends BaseRestHandler { @Inject public RestPutMappingAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(PUT, "/{index}/_mapping/", this); controller.registerHandler(PUT, "/{index}/{type}/_mapping", this); controller.registerHandler(PUT, "/{index}/_mapping/{type}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/open/RestOpenIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/open/RestOpenIndexAction.java index cb22f81ba4..58bda9d3a3 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/open/RestOpenIndexAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/open/RestOpenIndexAction.java @@ -39,7 +39,7 @@ public class RestOpenIndexAction extends BaseRestHandler { @Inject public RestOpenIndexAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.POST, "/_open", this); controller.registerHandler(RestRequest.Method.POST, "/{index}/_open", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/recovery/RestRecoveryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/recovery/RestRecoveryAction.java index e46831e81e..88bc9fb8c9 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/recovery/RestRecoveryAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/recovery/RestRecoveryAction.java @@ -45,7 +45,7 @@ public class RestRecoveryAction extends BaseRestHandler { @Inject public RestRecoveryAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_recovery", this); controller.registerHandler(GET, "/{index}/_recovery", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/refresh/RestRefreshAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/refresh/RestRefreshAction.java index e552b13316..fcc6d240b3 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/refresh/RestRefreshAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/refresh/RestRefreshAction.java @@ -47,7 +47,7 @@ public class RestRefreshAction extends BaseRestHandler { @Inject public RestRefreshAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_refresh", this); controller.registerHandler(POST, "/{index}/_refresh", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/segments/RestIndicesSegmentsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/segments/RestIndicesSegmentsAction.java index a233c75da5..da76a769ce 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/segments/RestIndicesSegmentsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/segments/RestIndicesSegmentsAction.java @@ -45,7 +45,7 @@ public class RestIndicesSegmentsAction extends BaseRestHandler { @Inject public RestIndicesSegmentsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_segments", this); controller.registerHandler(GET, "/{index}/_segments", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestGetSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestGetSettingsAction.java index f27897aa73..d41e09141a 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestGetSettingsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestGetSettingsAction.java @@ -44,7 +44,7 @@ public class RestGetSettingsAction extends BaseRestHandler { @Inject public RestGetSettingsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}/_settings/{name}", this); controller.registerHandler(GET, "/_settings/{name}", this); controller.registerHandler(GET, "/{index}/_setting/{name}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestUpdateSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestUpdateSettingsAction.java index 1a8ba58306..bcf43a4baa 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestUpdateSettingsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/settings/RestUpdateSettingsAction.java @@ -53,7 +53,7 @@ public class RestUpdateSettingsAction extends BaseRestHandler { @Inject public RestUpdateSettingsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.PUT, "/{index}/_settings", this); controller.registerHandler(RestRequest.Method.PUT, "/_settings", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/shards/RestIndicesShardStoresAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/shards/RestIndicesShardStoresAction.java index e2dc64cc47..586599c1a1 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/shards/RestIndicesShardStoresAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/shards/RestIndicesShardStoresAction.java @@ -46,7 +46,7 @@ public class RestIndicesShardStoresAction extends BaseRestHandler { @Inject public RestIndicesShardStoresAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_shard_stores", this); controller.registerHandler(GET, "/{index}/_shard_stores", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/stats/RestIndicesStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/stats/RestIndicesStatsAction.java index 891afd6b8c..e75dfcc4dc 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/stats/RestIndicesStatsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/stats/RestIndicesStatsAction.java @@ -47,7 +47,7 @@ public class RestIndicesStatsAction extends BaseRestHandler { @Inject public RestIndicesStatsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_stats", this); controller.registerHandler(GET, "/_stats/{metric}", this); controller.registerHandler(GET, "/_stats/{metric}/{indexMetric}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/delete/RestDeleteIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/delete/RestDeleteIndexTemplateAction.java index a4c1869609..a59ab9ac70 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/delete/RestDeleteIndexTemplateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/delete/RestDeleteIndexTemplateAction.java @@ -36,7 +36,7 @@ public class RestDeleteIndexTemplateAction extends BaseRestHandler { @Inject public RestDeleteIndexTemplateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.DELETE, "/_template/{name}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/get/RestGetIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/get/RestGetIndexTemplateAction.java index d5bfa0db90..d62d97400c 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/get/RestGetIndexTemplateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/get/RestGetIndexTemplateAction.java @@ -50,7 +50,7 @@ public class RestGetIndexTemplateAction extends BaseRestHandler { @Inject public RestGetIndexTemplateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_template", this); controller.registerHandler(GET, "/_template/{name}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/head/RestHeadIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/head/RestHeadIndexTemplateAction.java index 0838fa887e..648d083e76 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/head/RestHeadIndexTemplateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/head/RestHeadIndexTemplateAction.java @@ -42,7 +42,7 @@ public class RestHeadIndexTemplateAction extends BaseRestHandler { @Inject public RestHeadIndexTemplateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(HEAD, "/_template/{name}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/put/RestPutIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/put/RestPutIndexTemplateAction.java index 45f8a674dd..0b08b64e89 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/put/RestPutIndexTemplateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/template/put/RestPutIndexTemplateAction.java @@ -36,7 +36,7 @@ public class RestPutIndexTemplateAction extends BaseRestHandler { @Inject public RestPutIndexTemplateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(RestRequest.Method.PUT, "/_template/{name}", this); controller.registerHandler(RestRequest.Method.POST, "/_template/{name}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/upgrade/RestUpgradeAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/upgrade/RestUpgradeAction.java index 6a554db60f..60a781f90b 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/upgrade/RestUpgradeAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/upgrade/RestUpgradeAction.java @@ -49,7 +49,7 @@ public class RestUpgradeAction extends BaseRestHandler { @Inject public RestUpgradeAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_upgrade", this); controller.registerHandler(POST, "/{index}/_upgrade", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/query/RestValidateQueryAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/query/RestValidateQueryAction.java index 81bdaf7536..86d6e9d608 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/query/RestValidateQueryAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/query/RestValidateQueryAction.java @@ -57,7 +57,7 @@ public class RestValidateQueryAction extends BaseRestHandler { @Inject public RestValidateQueryAction(Settings settings, RestController controller, Client client, IndicesQueriesRegistry indicesQueriesRegistry) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_validate/query", this); controller.registerHandler(POST, "/_validate/query", this); controller.registerHandler(GET, "/{index}/_validate/query", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/template/RestRenderSearchTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/template/RestRenderSearchTemplateAction.java index 5ebec7130d..f130865752 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/template/RestRenderSearchTemplateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/validate/template/RestRenderSearchTemplateAction.java @@ -52,7 +52,7 @@ public class RestRenderSearchTemplateAction extends BaseRestHandler { @Inject public RestRenderSearchTemplateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_render/template", this); controller.registerHandler(POST, "/_render/template", this); controller.registerHandler(GET, "/_render/template/{id}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/bulk/RestBulkAction.java b/core/src/main/java/org/elasticsearch/rest/action/bulk/RestBulkAction.java index 37ce03bac7..3c0f4440d4 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/bulk/RestBulkAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/bulk/RestBulkAction.java @@ -58,7 +58,7 @@ public class RestBulkAction extends BaseRestHandler { @Inject public RestBulkAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_bulk", this); controller.registerHandler(PUT, "/_bulk", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java index 895211a097..12393f5800 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java @@ -39,7 +39,7 @@ import static org.elasticsearch.rest.action.support.RestTable.pad; public abstract class AbstractCatAction extends BaseRestHandler { public AbstractCatAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); } protected abstract void doRequest(final RestRequest request, final RestChannel channel, final Client client); diff --git a/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java b/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java index 337684769f..23229540b9 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/cat/RestCatAction.java @@ -41,7 +41,7 @@ public class RestCatAction extends BaseRestHandler { @Inject public RestCatAction(Settings settings, RestController controller, Set<AbstractCatAction> catActions, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_cat", this); StringBuilder sb = new StringBuilder(); sb.append(CAT_NL); diff --git a/core/src/main/java/org/elasticsearch/rest/action/count/RestCountAction.java b/core/src/main/java/org/elasticsearch/rest/action/count/RestCountAction.java index 834b3d391b..c423f7a853 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/count/RestCountAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/count/RestCountAction.java @@ -54,7 +54,7 @@ public class RestCountAction extends BaseRestHandler { @Inject public RestCountAction(Settings settings, RestController controller, Client client, IndicesQueriesRegistry indicesQueriesRegistry) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_count", this); controller.registerHandler(GET, "/_count", this); controller.registerHandler(POST, "/{index}/_count", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/delete/RestDeleteAction.java b/core/src/main/java/org/elasticsearch/rest/action/delete/RestDeleteAction.java index 4336c9db2d..8e3449344c 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/delete/RestDeleteAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/delete/RestDeleteAction.java @@ -41,7 +41,7 @@ public class RestDeleteAction extends BaseRestHandler { @Inject public RestDeleteAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(DELETE, "/{index}/{type}/{id}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/explain/RestExplainAction.java b/core/src/main/java/org/elasticsearch/rest/action/explain/RestExplainAction.java index 0e472bb0bf..864cddc4ba 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/explain/RestExplainAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/explain/RestExplainAction.java @@ -58,7 +58,7 @@ public class RestExplainAction extends BaseRestHandler { @Inject public RestExplainAction(Settings settings, RestController controller, Client client, IndicesQueriesRegistry indicesQueriesRegistry) { - super(settings, controller, client); + super(settings, client); this.indicesQueriesRegistry = indicesQueriesRegistry; controller.registerHandler(GET, "/{index}/{type}/{id}/_explain", this); controller.registerHandler(POST, "/{index}/{type}/{id}/_explain", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/fieldstats/RestFieldStatsAction.java b/core/src/main/java/org/elasticsearch/rest/action/fieldstats/RestFieldStatsAction.java index c314c4325d..17b406c71e 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/fieldstats/RestFieldStatsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/fieldstats/RestFieldStatsAction.java @@ -50,7 +50,7 @@ public class RestFieldStatsAction extends BaseRestHandler { @Inject public RestFieldStatsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_field_stats", this); controller.registerHandler(POST, "/_field_stats", this); controller.registerHandler(GET, "/{index}/_field_stats", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java b/core/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java index e85eef4857..0f541bf7a9 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java @@ -48,7 +48,7 @@ public class RestGetAction extends BaseRestHandler { @Inject public RestGetAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}/{type}/{id}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/get/RestGetSourceAction.java b/core/src/main/java/org/elasticsearch/rest/action/get/RestGetSourceAction.java index ff6c04a6d1..d38ad458c4 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/get/RestGetSourceAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/get/RestGetSourceAction.java @@ -48,7 +48,7 @@ public class RestGetSourceAction extends BaseRestHandler { @Inject public RestGetSourceAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}/{type}/{id}/_source", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/get/RestHeadAction.java b/core/src/main/java/org/elasticsearch/rest/action/get/RestHeadAction.java index f32c07f20f..31fd0cc9ea 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/get/RestHeadAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/get/RestHeadAction.java @@ -44,7 +44,7 @@ public class RestHeadAction extends BaseRestHandler { @Inject public RestHeadAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(HEAD, "/{index}/{type}/{id}", this); controller.registerHandler(HEAD, "/{index}/{type}/{id}/_source", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/get/RestMultiGetAction.java b/core/src/main/java/org/elasticsearch/rest/action/get/RestMultiGetAction.java index 440312b7cb..01a9c1b4e5 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/get/RestMultiGetAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/get/RestMultiGetAction.java @@ -42,7 +42,7 @@ public class RestMultiGetAction extends BaseRestHandler { @Inject public RestMultiGetAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_mget", this); controller.registerHandler(POST, "/_mget", this); controller.registerHandler(GET, "/{index}/_mget", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/index/RestIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/index/RestIndexAction.java index 13a9329918..5a65699df8 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/index/RestIndexAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/index/RestIndexAction.java @@ -47,7 +47,7 @@ public class RestIndexAction extends BaseRestHandler { @Inject public RestIndexAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/{index}/{type}", this); // auto id creation controller.registerHandler(PUT, "/{index}/{type}/{id}", this); controller.registerHandler(POST, "/{index}/{type}/{id}", this); @@ -58,7 +58,7 @@ public class RestIndexAction extends BaseRestHandler { final class CreateHandler extends BaseRestHandler { protected CreateHandler(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); } @Override diff --git a/core/src/main/java/org/elasticsearch/rest/action/main/RestMainAction.java b/core/src/main/java/org/elasticsearch/rest/action/main/RestMainAction.java index 42de9b898a..aaf0906b0f 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/main/RestMainAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/main/RestMainAction.java @@ -48,7 +48,7 @@ public class RestMainAction extends BaseRestHandler { @Inject public RestMainAction(Settings settings, Version version, RestController controller, ClusterName clusterName, Client client, ClusterService clusterService) { - super(settings, controller, client); + super(settings, client); this.version = version; this.clusterName = clusterName; this.clusterService = clusterService; diff --git a/core/src/main/java/org/elasticsearch/rest/action/percolate/RestMultiPercolateAction.java b/core/src/main/java/org/elasticsearch/rest/action/percolate/RestMultiPercolateAction.java index 879ec78d75..9e925022cf 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/percolate/RestMultiPercolateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/percolate/RestMultiPercolateAction.java @@ -44,7 +44,7 @@ public class RestMultiPercolateAction extends BaseRestHandler { @Inject public RestMultiPercolateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_mpercolate", this); controller.registerHandler(POST, "/{index}/_mpercolate", this); controller.registerHandler(POST, "/{index}/{type}/_mpercolate", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/percolate/RestPercolateAction.java b/core/src/main/java/org/elasticsearch/rest/action/percolate/RestPercolateAction.java index 052fa42104..a7c66b245e 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/percolate/RestPercolateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/percolate/RestPercolateAction.java @@ -44,7 +44,7 @@ public class RestPercolateAction extends BaseRestHandler { @Inject public RestPercolateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}/{type}/_percolate", this); controller.registerHandler(POST, "/{index}/{type}/_percolate", this); @@ -109,7 +109,7 @@ public class RestPercolateAction extends BaseRestHandler { final class RestCountPercolateDocHandler extends BaseRestHandler { private RestCountPercolateDocHandler(Settings settings, final RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); } @Override @@ -123,7 +123,7 @@ public class RestPercolateAction extends BaseRestHandler { final class RestPercolateExistingDocHandler extends BaseRestHandler { protected RestPercolateExistingDocHandler(Settings settings, final RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); } @Override @@ -136,7 +136,7 @@ public class RestPercolateAction extends BaseRestHandler { final class RestCountPercolateExistingDocHandler extends BaseRestHandler { protected RestCountPercolateExistingDocHandler(Settings settings, final RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); } @Override diff --git a/core/src/main/java/org/elasticsearch/rest/action/script/RestDeleteIndexedScriptAction.java b/core/src/main/java/org/elasticsearch/rest/action/script/RestDeleteIndexedScriptAction.java index b492e7c513..9009025d3a 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/script/RestDeleteIndexedScriptAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/script/RestDeleteIndexedScriptAction.java @@ -47,7 +47,7 @@ public class RestDeleteIndexedScriptAction extends BaseRestHandler { } protected RestDeleteIndexedScriptAction(Settings settings, RestController controller, boolean registerDefaultHandlers, Client client) { - super(settings, controller, client); + super(settings, client); if (registerDefaultHandlers) { controller.registerHandler(DELETE, "/_scripts/{lang}/{id}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/script/RestGetIndexedScriptAction.java b/core/src/main/java/org/elasticsearch/rest/action/script/RestGetIndexedScriptAction.java index a4c6784d41..e2c4ff6373 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/script/RestGetIndexedScriptAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/script/RestGetIndexedScriptAction.java @@ -48,7 +48,7 @@ public class RestGetIndexedScriptAction extends BaseRestHandler { } protected RestGetIndexedScriptAction(Settings settings, RestController controller, boolean registerDefaultHandlers, Client client) { - super(settings, controller, client); + super(settings, client); if (registerDefaultHandlers) { controller.registerHandler(GET, "/_scripts/{lang}/{id}", this); } diff --git a/core/src/main/java/org/elasticsearch/rest/action/script/RestPutIndexedScriptAction.java b/core/src/main/java/org/elasticsearch/rest/action/script/RestPutIndexedScriptAction.java index ed440c2b9f..f5a6f67517 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/script/RestPutIndexedScriptAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/script/RestPutIndexedScriptAction.java @@ -55,7 +55,7 @@ public class RestPutIndexedScriptAction extends BaseRestHandler { } protected RestPutIndexedScriptAction(Settings settings, RestController controller, boolean registerDefaultHandlers, Client client) { - super(settings, controller, client); + super(settings, client); if (registerDefaultHandlers) { controller.registerHandler(POST, "/_scripts/{lang}/{id}", this); controller.registerHandler(PUT, "/_scripts/{lang}/{id}", this); @@ -67,7 +67,7 @@ public class RestPutIndexedScriptAction extends BaseRestHandler { final class CreateHandler extends BaseRestHandler { protected CreateHandler(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); } @Override diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java index b2a2905585..0dce23bf3b 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java @@ -47,7 +47,7 @@ public class RestClearScrollAction extends BaseRestHandler { @Inject public RestClearScrollAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(DELETE, "/_search/scroll", this); controller.registerHandler(DELETE, "/_search/scroll/{scroll_id}", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java index ff51263e08..ed69dd6287 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java @@ -62,7 +62,7 @@ public class RestMultiSearchAction extends BaseRestHandler { @Inject public RestMultiSearchAction(Settings settings, RestController controller, Client client, IndicesQueriesRegistry indicesQueriesRegistry) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_msearch", this); controller.registerHandler(POST, "/_msearch", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java index 6db9531af8..e58caea532 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java @@ -65,7 +65,7 @@ public class RestSearchAction extends BaseRestHandler { @Inject public RestSearchAction(Settings settings, RestController controller, Client client, IndicesQueriesRegistry queryRegistry) { - super(settings, controller, client); + super(settings, client); this.queryRegistry = queryRegistry; controller.registerHandler(GET, "/_search", this); controller.registerHandler(POST, "/_search", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java index eb7e046590..9e9964245e 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java @@ -51,7 +51,7 @@ public class RestSearchScrollAction extends BaseRestHandler { @Inject public RestSearchScrollAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_search/scroll", this); controller.registerHandler(POST, "/_search/scroll", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/suggest/RestSuggestAction.java b/core/src/main/java/org/elasticsearch/rest/action/suggest/RestSuggestAction.java index 2841bbe1fe..4e6b88b68b 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/suggest/RestSuggestAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/suggest/RestSuggestAction.java @@ -49,7 +49,7 @@ public class RestSuggestAction extends BaseRestHandler { @Inject public RestSuggestAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/_suggest", this); controller.registerHandler(GET, "/_suggest", this); controller.registerHandler(POST, "/{index}/_suggest", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/template/RestPutSearchTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/template/RestPutSearchTemplateAction.java index 1523d299f0..4d0da8f0d1 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/template/RestPutSearchTemplateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/template/RestPutSearchTemplateAction.java @@ -50,7 +50,7 @@ public class RestPutSearchTemplateAction extends RestPutIndexedScriptAction { final class CreateHandler extends BaseRestHandler { protected CreateHandler(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); } @Override diff --git a/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestMultiTermVectorsAction.java b/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestMultiTermVectorsAction.java index fe897f9b09..dfcbeef171 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestMultiTermVectorsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestMultiTermVectorsAction.java @@ -40,7 +40,7 @@ public class RestMultiTermVectorsAction extends BaseRestHandler { @Inject public RestMultiTermVectorsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/_mtermvectors", this); controller.registerHandler(POST, "/_mtermvectors", this); controller.registerHandler(GET, "/{index}/_mtermvectors", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestTermVectorsAction.java b/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestTermVectorsAction.java index af81dfcd0a..dbbd885fe6 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestTermVectorsAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/termvectors/RestTermVectorsAction.java @@ -49,7 +49,7 @@ public class RestTermVectorsAction extends BaseRestHandler { @Inject public RestTermVectorsAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(GET, "/{index}/{type}/_termvectors", this); controller.registerHandler(POST, "/{index}/{type}/_termvectors", this); controller.registerHandler(GET, "/{index}/{type}/{id}/_termvectors", this); diff --git a/core/src/main/java/org/elasticsearch/rest/action/update/RestUpdateAction.java b/core/src/main/java/org/elasticsearch/rest/action/update/RestUpdateAction.java index 24264ca292..88f9037452 100644 --- a/core/src/main/java/org/elasticsearch/rest/action/update/RestUpdateAction.java +++ b/core/src/main/java/org/elasticsearch/rest/action/update/RestUpdateAction.java @@ -48,7 +48,7 @@ public class RestUpdateAction extends BaseRestHandler { @Inject public RestUpdateAction(Settings settings, RestController controller, Client client) { - super(settings, controller, client); + super(settings, client); controller.registerHandler(POST, "/{index}/{type}/{id}/_update", this); } diff --git a/core/src/main/java/org/elasticsearch/script/ScriptService.java b/core/src/main/java/org/elasticsearch/script/ScriptService.java index c9e9f9a873..37095513ae 100644 --- a/core/src/main/java/org/elasticsearch/script/ScriptService.java +++ b/core/src/main/java/org/elasticsearch/script/ScriptService.java @@ -31,7 +31,6 @@ import org.elasticsearch.action.indexedscripts.delete.DeleteIndexedScriptRequest import org.elasticsearch.action.indexedscripts.get.GetIndexedScriptRequest; import org.elasticsearch.action.indexedscripts.put.PutIndexedScriptRequest; import org.elasticsearch.client.Client; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.ParseField; import org.elasticsearch.common.ParseFieldMatcher; import org.elasticsearch.common.Strings; @@ -225,7 +224,7 @@ public class ScriptService extends AbstractComponent implements Closeable { /** * Checks if a script can be executed and compiles it if needed, or returns the previously compiled and cached script. */ - public CompiledScript compile(Script script, ScriptContext scriptContext, HasContextAndHeaders headersContext, Map<String, String> params) { + public CompiledScript compile(Script script, ScriptContext scriptContext, Map<String, String> params) { if (script == null) { throw new IllegalArgumentException("The parameter script (Script) must not be null."); } @@ -253,14 +252,14 @@ public class ScriptService extends AbstractComponent implements Closeable { " operation [" + scriptContext.getKey() + "] and lang [" + lang + "] are not supported"); } - return compileInternal(script, headersContext, params); + return compileInternal(script, params); } /** * Compiles a script straight-away, or returns the previously compiled and cached script, * without checking if it can be executed based on settings. */ - public CompiledScript compileInternal(Script script, HasContextAndHeaders context, Map<String, String> params) { + public CompiledScript compileInternal(Script script, Map<String, String> params) { if (script == null) { throw new IllegalArgumentException("The parameter script (Script) must not be null."); } @@ -297,7 +296,7 @@ public class ScriptService extends AbstractComponent implements Closeable { //the script has been updated in the index since the last look up. final IndexedScript indexedScript = new IndexedScript(lang, name); name = indexedScript.id; - code = getScriptFromIndex(indexedScript.lang, indexedScript.id, context); + code = getScriptFromIndex(indexedScript.lang, indexedScript.id); } CacheKey cacheKey = new CacheKey(scriptEngineService, type == ScriptType.INLINE ? null : name, code, params); @@ -323,7 +322,7 @@ public class ScriptService extends AbstractComponent implements Closeable { public void queryScriptIndex(GetIndexedScriptRequest request, final ActionListener<GetResponse> listener) { String scriptLang = validateScriptLanguage(request.scriptLang()); - GetRequest getRequest = new GetRequest(request, SCRIPT_INDEX).type(scriptLang).id(request.id()) + GetRequest getRequest = new GetRequest(SCRIPT_INDEX).type(scriptLang).id(request.id()) .version(request.version()).versionType(request.versionType()) .preference("_local"); //Set preference for no forking client.get(getRequest, listener); @@ -338,13 +337,12 @@ public class ScriptService extends AbstractComponent implements Closeable { return scriptLang; } - String getScriptFromIndex(String scriptLang, String id, HasContextAndHeaders context) { + String getScriptFromIndex(String scriptLang, String id) { if (client == null) { throw new IllegalArgumentException("Got an indexed script with no Client registered."); } scriptLang = validateScriptLanguage(scriptLang); GetRequest getRequest = new GetRequest(SCRIPT_INDEX, scriptLang, id); - getRequest.copyContextAndHeadersFrom(context); GetResponse responseFields = client.get(getRequest).actionGet(); if (responseFields.isExists()) { return getScriptFromResponse(responseFields); @@ -392,7 +390,7 @@ public class ScriptService extends AbstractComponent implements Closeable { //verify that the script compiles validate(request.source(), scriptLang); - IndexRequest indexRequest = new IndexRequest(request).index(SCRIPT_INDEX).type(scriptLang).id(request.id()) + IndexRequest indexRequest = new IndexRequest().index(SCRIPT_INDEX).type(scriptLang).id(request.id()) .version(request.version()).versionType(request.versionType()) .source(request.source()).opType(request.opType()).refresh(true); //Always refresh after indexing a template client.index(indexRequest, listener); @@ -400,7 +398,7 @@ public class ScriptService extends AbstractComponent implements Closeable { public void deleteScriptFromIndex(DeleteIndexedScriptRequest request, ActionListener<DeleteResponse> listener) { String scriptLang = validateScriptLanguage(request.scriptLang()); - DeleteRequest deleteRequest = new DeleteRequest(request).index(SCRIPT_INDEX).type(scriptLang).id(request.id()) + DeleteRequest deleteRequest = new DeleteRequest().index(SCRIPT_INDEX).type(scriptLang).id(request.id()) .refresh(true).version(request.version()).versionType(request.versionType()); client.delete(deleteRequest, listener); } @@ -437,8 +435,8 @@ public class ScriptService extends AbstractComponent implements Closeable { /** * Compiles (or retrieves from cache) and executes the provided script */ - public ExecutableScript executable(Script script, ScriptContext scriptContext, HasContextAndHeaders headersContext, Map<String, String> params) { - return executable(compile(script, scriptContext, headersContext, params), script.getParams()); + public ExecutableScript executable(Script script, ScriptContext scriptContext, Map<String, String> params) { + return executable(compile(script, scriptContext, params), script.getParams()); } /** @@ -452,7 +450,7 @@ public class ScriptService extends AbstractComponent implements Closeable { * Compiles (or retrieves from cache) and executes the provided search script */ public SearchScript search(SearchLookup lookup, Script script, ScriptContext scriptContext, Map<String, String> params) { - CompiledScript compiledScript = compile(script, scriptContext, SearchContext.current(), params); + CompiledScript compiledScript = compile(script, scriptContext, params); return getScriptEngineServiceForLang(compiledScript.lang()).search(compiledScript, lookup, script.getParams()); } diff --git a/core/src/main/java/org/elasticsearch/search/SearchService.java b/core/src/main/java/org/elasticsearch/search/SearchService.java index 473282a577..4737bd5cc3 100644 --- a/core/src/main/java/org/elasticsearch/search/SearchService.java +++ b/core/src/main/java/org/elasticsearch/search/SearchService.java @@ -566,7 +566,7 @@ public class SearchService extends AbstractLifecycleComponent<SearchService> imp context.scrollContext().scroll = request.scroll(); } if (request.template() != null) { - ExecutableScript executable = this.scriptService.executable(request.template(), ScriptContext.Standard.SEARCH, context, Collections.emptyMap()); + ExecutableScript executable = this.scriptService.executable(request.template(), ScriptContext.Standard.SEARCH, Collections.emptyMap()); BytesReference run = (BytesReference) executable.run(); try (XContentParser parser = XContentFactory.xContent(run).createParser(run)) { QueryParseContext queryParseContext = new QueryParseContext(indicesService.getIndicesQueryRegistry()); diff --git a/core/src/main/java/org/elasticsearch/search/action/SearchServiceTransportAction.java b/core/src/main/java/org/elasticsearch/search/action/SearchServiceTransportAction.java index 6e2bdf932f..138b215e68 100644 --- a/core/src/main/java/org/elasticsearch/search/action/SearchServiceTransportAction.java +++ b/core/src/main/java/org/elasticsearch/search/action/SearchServiceTransportAction.java @@ -125,7 +125,7 @@ public class SearchServiceTransportAction extends AbstractComponent { } public void sendClearAllScrollContexts(DiscoveryNode node, ClearScrollRequest request, final ActionListener<TransportResponse> listener) { - transportService.sendRequest(node, CLEAR_SCROLL_CONTEXTS_ACTION_NAME, new ClearScrollContextsRequest(request), new ActionListenerResponseHandler<TransportResponse>(listener) { + transportService.sendRequest(node, CLEAR_SCROLL_CONTEXTS_ACTION_NAME, new ClearScrollContextsRequest(), new ActionListenerResponseHandler<TransportResponse>(listener) { @Override public TransportResponse newInstance() { return TransportResponse.Empty.INSTANCE; @@ -220,11 +220,10 @@ public class SearchServiceTransportAction extends AbstractComponent { } ScrollFreeContextRequest(ClearScrollRequest request, long id) { - this((TransportRequest) request, id); + this(id); } - private ScrollFreeContextRequest(TransportRequest request, long id) { - super(request); + private ScrollFreeContextRequest(long id) { this.id = id; } @@ -252,7 +251,7 @@ public class SearchServiceTransportAction extends AbstractComponent { } SearchFreeContextRequest(SearchRequest request, long id) { - super(request, id); + super(id); this.originalIndices = new OriginalIndices(request); } @@ -322,14 +321,6 @@ public class SearchServiceTransportAction extends AbstractComponent { } public static class ClearScrollContextsRequest extends TransportRequest { - - public ClearScrollContextsRequest() { - } - - ClearScrollContextsRequest(TransportRequest request) { - super(request); - } - } class ClearScrollContextsTransportHandler implements TransportRequestHandler<ClearScrollContextsRequest> { diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/InternalAggregation.java b/core/src/main/java/org/elasticsearch/search/aggregations/InternalAggregation.java index 1c67a941da..04b1026446 100644 --- a/core/src/main/java/org/elasticsearch/search/aggregations/InternalAggregation.java +++ b/core/src/main/java/org/elasticsearch/search/aggregations/InternalAggregation.java @@ -18,8 +18,6 @@ */ package org.elasticsearch.search.aggregations; -import org.elasticsearch.common.DelegatingHasContextAndHeaders; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.bytes.BytesArray; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.io.stream.StreamInput; @@ -92,13 +90,12 @@ public abstract class InternalAggregation implements Aggregation, ToXContent, St } } - public static class ReduceContext extends DelegatingHasContextAndHeaders { + public static class ReduceContext { private final BigArrays bigArrays; private ScriptService scriptService; - public ReduceContext(BigArrays bigArrays, ScriptService scriptService, HasContextAndHeaders headersContext) { - super(headersContext); + public ReduceContext(BigArrays bigArrays, ScriptService scriptService) { this.bigArrays = bigArrays; this.scriptService = scriptService; } @@ -106,7 +103,7 @@ public abstract class InternalAggregation implements Aggregation, ToXContent, St public BigArrays bigArrays() { return bigArrays; } - + public ScriptService scriptService() { return scriptService; } diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/heuristics/ScriptHeuristic.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/heuristics/ScriptHeuristic.java index 9efea00051..a160451265 100644 --- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/heuristics/ScriptHeuristic.java +++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/heuristics/ScriptHeuristic.java @@ -87,7 +87,7 @@ public class ScriptHeuristic extends SignificanceHeuristic { @Override public void initialize(InternalAggregation.ReduceContext context) { - searchScript = context.scriptService().executable(script, ScriptContext.Standard.AGGS, context, Collections.emptyMap()); + searchScript = context.scriptService().executable(script, ScriptContext.Standard.AGGS, Collections.emptyMap()); searchScript.setNextVar("_subset_freq", subsetDfHolder); searchScript.setNextVar("_subset_size", subsetSizeHolder); searchScript.setNextVar("_superset_freq", supersetDfHolder); @@ -175,7 +175,7 @@ public class ScriptHeuristic extends SignificanceHeuristic { } ExecutableScript searchScript; try { - searchScript = scriptService.executable(script, ScriptContext.Standard.AGGS, context, Collections.emptyMap()); + searchScript = scriptService.executable(script, ScriptContext.Standard.AGGS, Collections.emptyMap()); } catch (Exception e) { throw new ElasticsearchParseException("failed to parse [{}] significance heuristic. the script [{}] could not be loaded", e, script, heuristicName); } diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/InternalScriptedMetric.java b/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/InternalScriptedMetric.java index 00c6b6b49b..3a516c690e 100644 --- a/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/InternalScriptedMetric.java +++ b/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/InternalScriptedMetric.java @@ -92,7 +92,7 @@ public class InternalScriptedMetric extends InternalMetricsAggregation implement vars.putAll(firstAggregation.reduceScript.getParams()); } CompiledScript compiledScript = reduceContext.scriptService().compile(firstAggregation.reduceScript, - ScriptContext.Standard.AGGS, reduceContext, Collections.emptyMap()); + ScriptContext.Standard.AGGS, Collections.emptyMap()); ExecutableScript script = reduceContext.scriptService().executable(compiledScript, vars); aggregation = script.run(); } else { diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/ScriptedMetricAggregator.java b/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/ScriptedMetricAggregator.java index 6603c6289b..68d886a7bf 100644 --- a/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/ScriptedMetricAggregator.java +++ b/core/src/main/java/org/elasticsearch/search/aggregations/metrics/scripted/ScriptedMetricAggregator.java @@ -59,11 +59,11 @@ public class ScriptedMetricAggregator extends MetricsAggregator { this.params = params; ScriptService scriptService = context.searchContext().scriptService(); if (initScript != null) { - scriptService.executable(initScript, ScriptContext.Standard.AGGS, context.searchContext(), Collections.emptyMap()).run(); + scriptService.executable(initScript, ScriptContext.Standard.AGGS, Collections.emptyMap()).run(); } this.mapScript = scriptService.search(context.searchContext().lookup(), mapScript, ScriptContext.Standard.AGGS, Collections.emptyMap()); if (combineScript != null) { - this.combineScript = scriptService.executable(combineScript, ScriptContext.Standard.AGGS, context.searchContext(), Collections.emptyMap()); + this.combineScript = scriptService.executable(combineScript, ScriptContext.Standard.AGGS, Collections.emptyMap()); } else { this.combineScript = null; } diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketscript/BucketScriptPipelineAggregator.java b/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketscript/BucketScriptPipelineAggregator.java index 76cb15ed46..4da355f4f8 100644 --- a/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketscript/BucketScriptPipelineAggregator.java +++ b/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketscript/BucketScriptPipelineAggregator.java @@ -94,7 +94,7 @@ public class BucketScriptPipelineAggregator extends PipelineAggregator { InternalMultiBucketAggregation<InternalMultiBucketAggregation, InternalMultiBucketAggregation.InternalBucket> originalAgg = (InternalMultiBucketAggregation<InternalMultiBucketAggregation, InternalMultiBucketAggregation.InternalBucket>) aggregation; List<? extends Bucket> buckets = originalAgg.getBuckets(); - CompiledScript compiledScript = reduceContext.scriptService().compile(script, ScriptContext.Standard.AGGS, reduceContext, Collections.emptyMap()); + CompiledScript compiledScript = reduceContext.scriptService().compile(script, ScriptContext.Standard.AGGS, Collections.emptyMap()); List newBuckets = new ArrayList<>(); for (Bucket bucket : buckets) { Map<String, Object> vars = new HashMap<>(); diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/having/BucketSelectorPipelineAggregator.java b/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/having/BucketSelectorPipelineAggregator.java index edc3b4e87c..1032d0f017 100644 --- a/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/having/BucketSelectorPipelineAggregator.java +++ b/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/having/BucketSelectorPipelineAggregator.java @@ -89,7 +89,7 @@ public class BucketSelectorPipelineAggregator extends PipelineAggregator { InternalMultiBucketAggregation<InternalMultiBucketAggregation, InternalMultiBucketAggregation.InternalBucket> originalAgg = (InternalMultiBucketAggregation<InternalMultiBucketAggregation, InternalMultiBucketAggregation.InternalBucket>) aggregation; List<? extends Bucket> buckets = originalAgg.getBuckets(); - CompiledScript compiledScript = reduceContext.scriptService().compile(script, ScriptContext.Standard.AGGS, reduceContext, Collections.emptyMap()); + CompiledScript compiledScript = reduceContext.scriptService().compile(script, ScriptContext.Standard.AGGS, Collections.emptyMap()); List newBuckets = new ArrayList<>(); for (Bucket bucket : buckets) { Map<String, Object> vars = new HashMap<>(); diff --git a/core/src/main/java/org/elasticsearch/search/controller/SearchPhaseController.java b/core/src/main/java/org/elasticsearch/search/controller/SearchPhaseController.java index ef16a03831..d79b1f59a6 100644 --- a/core/src/main/java/org/elasticsearch/search/controller/SearchPhaseController.java +++ b/core/src/main/java/org/elasticsearch/search/controller/SearchPhaseController.java @@ -31,7 +31,6 @@ import org.apache.lucene.search.TermStatistics; import org.apache.lucene.search.TopDocs; import org.apache.lucene.search.TopFieldDocs; import org.elasticsearch.action.search.SearchRequest; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.collect.HppcMaps; import org.elasticsearch.common.component.AbstractComponent; import org.elasticsearch.common.inject.Inject; @@ -299,7 +298,7 @@ public class SearchPhaseController extends AbstractComponent { } public InternalSearchResponse merge(ScoreDoc[] sortedDocs, AtomicArray<? extends QuerySearchResultProvider> queryResultsArr, - AtomicArray<? extends FetchSearchResultProvider> fetchResultsArr, HasContextAndHeaders headersContext) { + AtomicArray<? extends FetchSearchResultProvider> fetchResultsArr) { List<? extends AtomicArray.Entry<? extends QuerySearchResultProvider>> queryResults = queryResultsArr.asList(); List<? extends AtomicArray.Entry<? extends FetchSearchResultProvider>> fetchResults = fetchResultsArr.asList(); @@ -407,7 +406,7 @@ public class SearchPhaseController extends AbstractComponent { for (AtomicArray.Entry<? extends QuerySearchResultProvider> entry : queryResults) { aggregationsList.add((InternalAggregations) entry.value.queryResult().aggregations()); } - aggregations = InternalAggregations.reduce(aggregationsList, new ReduceContext(bigArrays, scriptService, headersContext)); + aggregations = InternalAggregations.reduce(aggregationsList, new ReduceContext(bigArrays, scriptService)); } } @@ -430,7 +429,7 @@ public class SearchPhaseController extends AbstractComponent { }).collect(Collectors.toList()); for (SiblingPipelineAggregator pipelineAggregator : pipelineAggregators) { InternalAggregation newAgg = pipelineAggregator.doReduce(new InternalAggregations(newAggs), new ReduceContext( - bigArrays, scriptService, headersContext)); + bigArrays, scriptService)); newAggs.add(newAgg); } aggregations = new InternalAggregations(newAggs); diff --git a/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java b/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java index 0d524ed3e3..4087eb9a01 100644 --- a/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java +++ b/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java @@ -22,7 +22,6 @@ package org.elasticsearch.search.fetch; import com.carrotsearch.hppc.IntArrayList; import org.apache.lucene.search.FieldDoc; import org.apache.lucene.search.ScoreDoc; -import org.elasticsearch.action.search.SearchScrollRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.lucene.Lucene; @@ -47,16 +46,7 @@ public class ShardFetchRequest extends TransportRequest { public ShardFetchRequest() { } - public ShardFetchRequest(SearchScrollRequest request, long id, IntArrayList list, ScoreDoc lastEmittedDoc) { - super(request); - this.id = id; - this.docIds = list.buffer; - this.size = list.size(); - this.lastEmittedDoc = lastEmittedDoc; - } - - protected ShardFetchRequest(TransportRequest originalRequest, long id, IntArrayList list, ScoreDoc lastEmittedDoc) { - super(originalRequest); + public ShardFetchRequest(long id, IntArrayList list, ScoreDoc lastEmittedDoc) { this.id = id; this.docIds = list.buffer; this.size = list.size(); diff --git a/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchSearchRequest.java b/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchSearchRequest.java index cc53b48f13..d908aca0fc 100644 --- a/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchSearchRequest.java +++ b/core/src/main/java/org/elasticsearch/search/fetch/ShardFetchSearchRequest.java @@ -46,7 +46,7 @@ public class ShardFetchSearchRequest extends ShardFetchRequest implements Indice } public ShardFetchSearchRequest(SearchRequest request, long id, IntArrayList list, ScoreDoc lastEmittedDoc) { - super(request, id, list, lastEmittedDoc); + super(id, list, lastEmittedDoc); this.originalIndices = new OriginalIndices(request); } diff --git a/core/src/main/java/org/elasticsearch/search/internal/DefaultSearchContext.java b/core/src/main/java/org/elasticsearch/search/internal/DefaultSearchContext.java index 3b7848d7d2..e935dc2575 100644 --- a/core/src/main/java/org/elasticsearch/search/internal/DefaultSearchContext.java +++ b/core/src/main/java/org/elasticsearch/search/internal/DefaultSearchContext.java @@ -165,7 +165,7 @@ public class DefaultSearchContext extends SearchContext { BigArrays bigArrays, Counter timeEstimateCounter, ParseFieldMatcher parseFieldMatcher, TimeValue timeout ) { - super(parseFieldMatcher, request); + super(parseFieldMatcher); this.id = id; this.request = request; this.searchType = request.searchType(); diff --git a/core/src/main/java/org/elasticsearch/search/internal/FilteredSearchContext.java b/core/src/main/java/org/elasticsearch/search/internal/FilteredSearchContext.java index eaa14933b3..72d923eb62 100644 --- a/core/src/main/java/org/elasticsearch/search/internal/FilteredSearchContext.java +++ b/core/src/main/java/org/elasticsearch/search/internal/FilteredSearchContext.java @@ -62,7 +62,7 @@ public abstract class FilteredSearchContext extends SearchContext { public FilteredSearchContext(SearchContext in) { //inner_hits in percolator ends up with null inner search context - super(in == null ? ParseFieldMatcher.EMPTY : in.parseFieldMatcher(), in); + super(in == null ? ParseFieldMatcher.EMPTY : in.parseFieldMatcher()); this.in = in; } diff --git a/core/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java b/core/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java index 77a490af38..7f91813804 100644 --- a/core/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java +++ b/core/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java @@ -42,7 +42,6 @@ public class InternalScrollSearchRequest extends TransportRequest { } public InternalScrollSearchRequest(SearchScrollRequest request, long id) { - super(request); this.id = id; this.scroll = request.scroll(); } diff --git a/core/src/main/java/org/elasticsearch/search/internal/SearchContext.java b/core/src/main/java/org/elasticsearch/search/internal/SearchContext.java index 76164b5c0f..822031c1a0 100644 --- a/core/src/main/java/org/elasticsearch/search/internal/SearchContext.java +++ b/core/src/main/java/org/elasticsearch/search/internal/SearchContext.java @@ -25,8 +25,6 @@ import org.apache.lucene.search.Sort; import org.apache.lucene.util.Counter; import org.elasticsearch.action.search.SearchType; import org.elasticsearch.cache.recycler.PageCacheRecycler; -import org.elasticsearch.common.DelegatingHasContextAndHeaders; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.ParseFieldMatcher; import org.elasticsearch.common.lease.Releasable; @@ -66,7 +64,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; -public abstract class SearchContext extends DelegatingHasContextAndHeaders implements Releasable { +public abstract class SearchContext implements Releasable { private static ThreadLocal<SearchContext> current = new ThreadLocal<>(); public final static int DEFAULT_TERMINATE_AFTER = 0; @@ -90,8 +88,7 @@ public abstract class SearchContext extends DelegatingHasContextAndHeaders imple protected final ParseFieldMatcher parseFieldMatcher; - protected SearchContext(ParseFieldMatcher parseFieldMatcher, HasContextAndHeaders contextHeaders) { - super(contextHeaders); + protected SearchContext(ParseFieldMatcher parseFieldMatcher) { this.parseFieldMatcher = parseFieldMatcher; } diff --git a/core/src/main/java/org/elasticsearch/search/internal/ShardSearchLocalRequest.java b/core/src/main/java/org/elasticsearch/search/internal/ShardSearchLocalRequest.java index 9d15dfd579..4a42f77f82 100644 --- a/core/src/main/java/org/elasticsearch/search/internal/ShardSearchLocalRequest.java +++ b/core/src/main/java/org/elasticsearch/search/internal/ShardSearchLocalRequest.java @@ -22,7 +22,6 @@ package org.elasticsearch.search.internal; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchType; import org.elasticsearch.cluster.routing.ShardRouting; -import org.elasticsearch.common.ContextAndHeaderHolder; import org.elasticsearch.common.Strings; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.io.stream.BytesStreamOutput; @@ -57,7 +56,7 @@ import static org.elasticsearch.search.Scroll.readScroll; * </pre> */ -public class ShardSearchLocalRequest extends ContextAndHeaderHolder implements ShardSearchRequest { +public class ShardSearchLocalRequest implements ShardSearchRequest { private String index; private int shardId; @@ -84,7 +83,6 @@ public class ShardSearchLocalRequest extends ContextAndHeaderHolder implements S this.scroll = searchRequest.scroll(); this.filteringAliases = filteringAliases; this.nowInMillis = nowInMillis; - copyContextAndHeadersFrom(searchRequest); } public ShardSearchLocalRequest(String[] types, long nowInMillis) { diff --git a/core/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java b/core/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java index b1730b6a14..1f0b3d1f18 100644 --- a/core/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java +++ b/core/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java @@ -20,7 +20,6 @@ package org.elasticsearch.search.internal; import org.elasticsearch.action.search.SearchType; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.script.Template; import org.elasticsearch.search.Scroll; @@ -33,7 +32,7 @@ import java.io.IOException; * It provides all the methods that the {@link org.elasticsearch.search.internal.SearchContext} needs. * Provides a cache key based on its content that can be used to cache shard level response. */ -public interface ShardSearchRequest extends HasContextAndHeaders { +public interface ShardSearchRequest { String index(); diff --git a/core/src/main/java/org/elasticsearch/search/internal/ShardSearchTransportRequest.java b/core/src/main/java/org/elasticsearch/search/internal/ShardSearchTransportRequest.java index 0f9c0ced41..48ea31c170 100644 --- a/core/src/main/java/org/elasticsearch/search/internal/ShardSearchTransportRequest.java +++ b/core/src/main/java/org/elasticsearch/search/internal/ShardSearchTransportRequest.java @@ -51,7 +51,6 @@ public class ShardSearchTransportRequest extends TransportRequest implements Sha public ShardSearchTransportRequest(SearchRequest searchRequest, ShardRouting shardRouting, int numberOfShards, String[] filteringAliases, long nowInMillis) { - super(searchRequest); this.shardSearchLocalRequest = new ShardSearchLocalRequest(searchRequest, shardRouting, numberOfShards, filteringAliases, nowInMillis); this.originalIndices = new OriginalIndices(searchRequest); } diff --git a/core/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java b/core/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java index a1395bd73d..15593abf0d 100644 --- a/core/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java +++ b/core/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java @@ -47,7 +47,6 @@ public class QuerySearchRequest extends TransportRequest implements IndicesReque } public QuerySearchRequest(SearchRequest request, long id, AggregatedDfs dfs) { - super(request); this.id = id; this.dfs = dfs; this.originalIndices = new OriginalIndices(request); diff --git a/core/src/main/java/org/elasticsearch/search/suggest/SuggestContextParser.java b/core/src/main/java/org/elasticsearch/search/suggest/SuggestContextParser.java index a8050d1aca..a7aa3fd60b 100644 --- a/core/src/main/java/org/elasticsearch/search/suggest/SuggestContextParser.java +++ b/core/src/main/java/org/elasticsearch/search/suggest/SuggestContextParser.java @@ -18,7 +18,6 @@ */ package org.elasticsearch.search.suggest; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.index.fielddata.IndexFieldDataService; import org.elasticsearch.index.mapper.MapperService; @@ -26,6 +25,6 @@ import org.elasticsearch.index.mapper.MapperService; import java.io.IOException; public interface SuggestContextParser { - SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService indexFieldDataService, HasContextAndHeaders headersContext) throws IOException; + SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService indexFieldDataService) throws IOException; -}
\ No newline at end of file +} diff --git a/core/src/main/java/org/elasticsearch/search/suggest/SuggestParseElement.java b/core/src/main/java/org/elasticsearch/search/suggest/SuggestParseElement.java index 650eb76b1c..a8a4e9ec26 100644 --- a/core/src/main/java/org/elasticsearch/search/suggest/SuggestParseElement.java +++ b/core/src/main/java/org/elasticsearch/search/suggest/SuggestParseElement.java @@ -19,7 +19,6 @@ package org.elasticsearch.search.suggest; import org.apache.lucene.util.BytesRef; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.inject.Inject; import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.index.fielddata.IndexFieldDataService; @@ -46,12 +45,12 @@ public final class SuggestParseElement implements SearchParseElement { @Override public void parse(XContentParser parser, SearchContext context) throws Exception { SuggestionSearchContext suggestionSearchContext = parseInternal(parser, context.mapperService(), context.fieldData(), - context.shardTarget().index(), context.shardTarget().shardId(), context); + context.shardTarget().index(), context.shardTarget().shardId()); context.suggest(suggestionSearchContext); } public SuggestionSearchContext parseInternal(XContentParser parser, MapperService mapperService, IndexFieldDataService fieldDataService, - String index, int shardId, HasContextAndHeaders headersContext) throws IOException { + String index, int shardId) throws IOException { SuggestionSearchContext suggestionSearchContext = new SuggestionSearchContext(); BytesRef globalText = null; @@ -96,7 +95,7 @@ public final class SuggestParseElement implements SearchParseElement { throw new IllegalArgumentException("Suggester[" + fieldName + "] not supported"); } final SuggestContextParser contextParser = suggesters.get(fieldName).getContextParser(); - suggestionContext = contextParser.parse(parser, mapperService, fieldDataService, headersContext); + suggestionContext = contextParser.parse(parser, mapperService, fieldDataService); } } if (suggestionContext != null) { diff --git a/core/src/main/java/org/elasticsearch/search/suggest/completion/CompletionSuggestParser.java b/core/src/main/java/org/elasticsearch/search/suggest/completion/CompletionSuggestParser.java index a2e5f743c5..7d886e1c7e 100644 --- a/core/src/main/java/org/elasticsearch/search/suggest/completion/CompletionSuggestParser.java +++ b/core/src/main/java/org/elasticsearch/search/suggest/completion/CompletionSuggestParser.java @@ -20,7 +20,6 @@ package org.elasticsearch.search.suggest.completion; import org.apache.lucene.analysis.Analyzer; import org.elasticsearch.ElasticsearchException; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.ParseField; import org.elasticsearch.common.bytes.BytesReference; import org.elasticsearch.common.unit.Fuzziness; @@ -134,8 +133,7 @@ public class CompletionSuggestParser implements SuggestContextParser { } @Override - public SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService fieldDataService, - HasContextAndHeaders headersContext) throws IOException { + public SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService fieldDataService) throws IOException { final CompletionSuggestionContext suggestion = new CompletionSuggestionContext(completionSuggester, mapperService, fieldDataService); final ContextAndSuggest contextAndSuggest = new ContextAndSuggest(mapperService); TLP_PARSER.parse(parser, suggestion, contextAndSuggest); diff --git a/core/src/main/java/org/elasticsearch/search/suggest/phrase/PhraseSuggestParser.java b/core/src/main/java/org/elasticsearch/search/suggest/phrase/PhraseSuggestParser.java index 0b904a9572..b4776652fb 100644 --- a/core/src/main/java/org/elasticsearch/search/suggest/phrase/PhraseSuggestParser.java +++ b/core/src/main/java/org/elasticsearch/search/suggest/phrase/PhraseSuggestParser.java @@ -22,7 +22,6 @@ import org.apache.lucene.analysis.Analyzer; import org.apache.lucene.index.IndexReader; import org.apache.lucene.index.Terms; import org.apache.lucene.util.BytesRef; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.ParseFieldMatcher; import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.common.xcontent.XContentParser.Token; @@ -50,8 +49,7 @@ public final class PhraseSuggestParser implements SuggestContextParser { } @Override - public SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService fieldDataService, - HasContextAndHeaders headersContext) throws IOException { + public SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService fieldDataService) throws IOException { PhraseSuggestionContext suggestion = new PhraseSuggestionContext(suggester); ParseFieldMatcher parseFieldMatcher = mapperService.getIndexSettings().getParseFieldMatcher(); XContentParser.Token token; @@ -143,8 +141,7 @@ public final class PhraseSuggestParser implements SuggestContextParser { throw new IllegalArgumentException("suggester[phrase][collate] query already set, doesn't support additional [" + fieldName + "]"); } Template template = Template.parse(parser, parseFieldMatcher); - CompiledScript compiledScript = suggester.scriptService().compile(template, ScriptContext.Standard.SEARCH, - headersContext, Collections.emptyMap()); + CompiledScript compiledScript = suggester.scriptService().compile(template, ScriptContext.Standard.SEARCH, Collections.emptyMap()); suggestion.setCollateQueryScript(compiledScript); } else if ("params".equals(fieldName)) { suggestion.setCollateScriptParams(parser.map()); diff --git a/core/src/main/java/org/elasticsearch/search/suggest/term/TermSuggestParser.java b/core/src/main/java/org/elasticsearch/search/suggest/term/TermSuggestParser.java index a0e0e28954..a2fd680c21 100644 --- a/core/src/main/java/org/elasticsearch/search/suggest/term/TermSuggestParser.java +++ b/core/src/main/java/org/elasticsearch/search/suggest/term/TermSuggestParser.java @@ -18,7 +18,6 @@ */ package org.elasticsearch.search.suggest.term; -import org.elasticsearch.common.HasContextAndHeaders; import org.elasticsearch.common.ParseFieldMatcher; import org.elasticsearch.common.xcontent.XContentParser; import org.elasticsearch.index.fielddata.IndexFieldDataService; @@ -39,8 +38,7 @@ public final class TermSuggestParser implements SuggestContextParser { } @Override - public SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService fieldDataService, - HasContextAndHeaders headersContext) throws IOException { + public SuggestionSearchContext.SuggestionContext parse(XContentParser parser, MapperService mapperService, IndexFieldDataService fieldDataService) throws IOException { XContentParser.Token token; String fieldName = null; TermSuggestionContext suggestion = new TermSuggestionContext(suggester); diff --git a/core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java b/core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java index 0e6204ddd1..378a849115 100644 --- a/core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java +++ b/core/src/main/java/org/elasticsearch/threadpool/ThreadPool.java @@ -20,6 +20,7 @@ package org.elasticsearch.threadpool; import org.apache.lucene.util.Counter; +import org.apache.lucene.util.IOUtils; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.component.AbstractComponent; import org.elasticsearch.common.io.stream.StreamInput; @@ -34,11 +35,13 @@ import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.util.concurrent.EsAbortPolicy; import org.elasticsearch.common.util.concurrent.EsExecutors; import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.common.util.concurrent.XRejectedExecutionHandler; import org.elasticsearch.common.xcontent.ToXContent; import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.common.xcontent.XContentBuilderString; +import java.io.Closeable; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; @@ -67,7 +70,7 @@ import static org.elasticsearch.common.unit.TimeValue.timeValueMinutes; /** * */ -public class ThreadPool extends AbstractComponent { +public class ThreadPool extends AbstractComponent implements Closeable { public static class Names { public static final String SAME = "same"; @@ -200,6 +203,8 @@ public class ThreadPool extends AbstractComponent { static final Executor DIRECT_EXECUTOR = command -> command.run(); + private final ThreadContext threadContext; + public ThreadPool(String name) { this(Settings.builder().put("name", name).build()); } @@ -208,7 +213,7 @@ public class ThreadPool extends AbstractComponent { super(settings); assert settings.get("name") != null : "ThreadPool's settings should contain a name"; - + threadContext = new ThreadContext(settings); Map<String, Settings> groupSettings = THREADPOOL_GROUP_SETTING.get(settings).getAsGroups(); validate(groupSettings); @@ -448,7 +453,7 @@ public class ThreadPool extends AbstractComponent { } else { logger.debug("creating thread_pool [{}], type [{}], keep_alive [{}]", name, type, keepAlive); } - Executor executor = EsExecutors.newCached(name, keepAlive.millis(), TimeUnit.MILLISECONDS, threadFactory); + Executor executor = EsExecutors.newCached(name, keepAlive.millis(), TimeUnit.MILLISECONDS, threadFactory, threadContext); return new ExecutorHolder(executor, new Info(name, threadPoolType, -1, -1, keepAlive, null)); } else if (ThreadPoolType.FIXED == threadPoolType) { int defaultSize = defaultSettings.getAsInt("size", EsExecutors.boundedNumberOfProcessors(settings)); @@ -483,7 +488,7 @@ public class ThreadPool extends AbstractComponent { int size = applyHardSizeLimit(name, settings.getAsInt("size", defaultSize)); SizeValue queueSize = getAsSizeOrUnbounded(settings, "capacity", getAsSizeOrUnbounded(settings, "queue", getAsSizeOrUnbounded(settings, "queue_size", defaultQueueSize))); logger.debug("creating thread_pool [{}], type [{}], size [{}], queue_size [{}]", name, type, size, queueSize); - Executor executor = EsExecutors.newFixed(name, size, queueSize == null ? -1 : (int) queueSize.singles(), threadFactory); + Executor executor = EsExecutors.newFixed(name, size, queueSize == null ? -1 : (int) queueSize.singles(), threadFactory, threadContext); return new ExecutorHolder(executor, new Info(name, threadPoolType, size, size, null, queueSize)); } else if (ThreadPoolType.SCALING == threadPoolType) { TimeValue defaultKeepAlive = defaultSettings.getAsTime("keep_alive", timeValueMinutes(5)); @@ -527,7 +532,7 @@ public class ThreadPool extends AbstractComponent { } else { logger.debug("creating thread_pool [{}], type [{}], min [{}], size [{}], keep_alive [{}]", name, type, min, size, keepAlive); } - Executor executor = EsExecutors.newScaling(name, min, size, keepAlive.millis(), TimeUnit.MILLISECONDS, threadFactory); + Executor executor = EsExecutors.newScaling(name, min, size, keepAlive.millis(), TimeUnit.MILLISECONDS, threadFactory, threadContext); return new ExecutorHolder(executor, new Info(name, threadPoolType, min, size, keepAlive, null)); } throw new IllegalArgumentException("No type found [" + type + "], for [" + name + "]"); @@ -914,17 +919,30 @@ public class ThreadPool extends AbstractComponent { */ public static boolean terminate(ThreadPool pool, long timeout, TimeUnit timeUnit) { if (pool != null) { - pool.shutdown(); try { - if (pool.awaitTermination(timeout, timeUnit)) { - return true; + pool.shutdown(); + try { + if (pool.awaitTermination(timeout, timeUnit)) { + return true; + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); + // last resort + pool.shutdownNow(); + } finally { + IOUtils.closeWhileHandlingException(pool); } - // last resort - pool.shutdownNow(); } return false; } + + @Override + public void close() throws IOException { + threadContext.close(); + } + + public ThreadContext getThreadContext() { + return threadContext; + } } diff --git a/core/src/main/java/org/elasticsearch/transport/TransportMessage.java b/core/src/main/java/org/elasticsearch/transport/TransportMessage.java index f52f9179dd..1434a6eaa7 100644 --- a/core/src/main/java/org/elasticsearch/transport/TransportMessage.java +++ b/core/src/main/java/org/elasticsearch/transport/TransportMessage.java @@ -19,29 +19,20 @@ package org.elasticsearch.transport; -import org.elasticsearch.common.ContextAndHeaderHolder; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Streamable; import org.elasticsearch.common.transport.TransportAddress; import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; -/** - * The transport message is also a {@link ContextAndHeaderHolder context holder} that holds <b>transient</b> context, that is, - * the context is not serialized with message. - */ -public abstract class TransportMessage<TM extends TransportMessage<TM>> extends ContextAndHeaderHolder implements Streamable { +public abstract class TransportMessage<TM extends TransportMessage<TM>> implements Streamable { private TransportAddress remoteAddress; - protected TransportMessage() { - } - - protected TransportMessage(TM message) { - copyContextAndHeadersFrom(message); - } - public void remoteAddress(TransportAddress remoteAddress) { this.remoteAddress = remoteAddress; } @@ -52,16 +43,11 @@ public abstract class TransportMessage<TM extends TransportMessage<TM>> extends @Override public void readFrom(StreamInput in) throws IOException { - headers = in.readBoolean() ? in.readMap() : null; + } @Override public void writeTo(StreamOutput out) throws IOException { - if (headers == null) { - out.writeBoolean(false); - } else { - out.writeBoolean(true); - out.writeMap(headers); - } + } } diff --git a/core/src/main/java/org/elasticsearch/transport/TransportRequest.java b/core/src/main/java/org/elasticsearch/transport/TransportRequest.java index d5c1491f1a..7db7f0726d 100644 --- a/core/src/main/java/org/elasticsearch/transport/TransportRequest.java +++ b/core/src/main/java/org/elasticsearch/transport/TransportRequest.java @@ -26,24 +26,12 @@ import org.elasticsearch.tasks.Task; public abstract class TransportRequest extends TransportMessage<TransportRequest> { public static class Empty extends TransportRequest { - public static final Empty INSTANCE = new Empty(); - - public Empty() { - super(); - } - - public Empty(TransportRequest request) { - super(request); - } } public TransportRequest() { } - protected TransportRequest(TransportRequest request) { - super(request); - } public Task createTask(long id, String type, String action) { return new Task(id, type, action, this::getDescription); diff --git a/core/src/main/java/org/elasticsearch/transport/TransportResponse.java b/core/src/main/java/org/elasticsearch/transport/TransportResponse.java index 8ea7cd60d2..28dcd12ca5 100644 --- a/core/src/main/java/org/elasticsearch/transport/TransportResponse.java +++ b/core/src/main/java/org/elasticsearch/transport/TransportResponse.java @@ -24,23 +24,6 @@ package org.elasticsearch.transport; public abstract class TransportResponse extends TransportMessage<TransportResponse> { public static class Empty extends TransportResponse { - public static final Empty INSTANCE = new Empty(); - - public Empty() { - super(); - } - - public Empty(TransportResponse request) { - super(request); - } - } - - protected TransportResponse() { - } - - protected TransportResponse(TransportResponse response) { - super(response); } - } diff --git a/core/src/main/java/org/elasticsearch/transport/TransportService.java b/core/src/main/java/org/elasticsearch/transport/TransportService.java index 5d74c4a408..8cff05a4d6 100644 --- a/core/src/main/java/org/elasticsearch/transport/TransportService.java +++ b/core/src/main/java/org/elasticsearch/transport/TransportService.java @@ -40,6 +40,7 @@ import org.elasticsearch.common.util.concurrent.ConcurrentMapLong; import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; import org.elasticsearch.common.util.concurrent.FutureUtils; import org.elasticsearch.tasks.TaskManager; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.threadpool.ThreadPool; import java.io.IOException; @@ -288,7 +289,7 @@ public class TransportService extends AbstractLifecycleComponent<TransportServic } else { timeoutHandler = new TimeoutHandler(requestId); } - clientHandlers.put(requestId, new RequestHolder<>(handler, node, action, timeoutHandler)); + clientHandlers.put(requestId, new RequestHolder<>(new ContextRestoreResponseHandler<T>(threadPool.getThreadContext().newStoredContext(), handler), node, action, timeoutHandler)); if (started.get() == false) { // if we are not started the exception handling will remove the RequestHolder again and calls the handler to notify the caller. // it will only notify if the toStop code hasn't done the work yet. @@ -494,6 +495,7 @@ public class TransportService extends AbstractLifecycleComponent<TransportServic @Override public TransportResponseHandler onResponseReceived(final long requestId) { RequestHolder holder = clientHandlers.remove(requestId); + if (holder == null) { checkForTimeout(requestId); return null; @@ -708,6 +710,41 @@ public class TransportService extends AbstractLifecycleComponent<TransportServic } } + /** + * This handler wrapper ensures that the response thread executes with the correct thread context. Before any of the4 handle methods + * are invoked we restore the context. + */ + private final static class ContextRestoreResponseHandler<T extends TransportResponse> implements TransportResponseHandler<T> { + private final TransportResponseHandler<T> delegate; + private final ThreadContext.StoredContext threadContext; + private ContextRestoreResponseHandler(ThreadContext.StoredContext threadContext, TransportResponseHandler<T> delegate) { + this.delegate = delegate; + this.threadContext = threadContext; + } + + @Override + public T newInstance() { + return delegate.newInstance(); + } + + @Override + public void handleResponse(T response) { + threadContext.restore(); + delegate.handleResponse(response); + } + + @Override + public void handleException(TransportException exp) { + threadContext.restore(); + delegate.handleException(exp); + } + + @Override + public String executor() { + return delegate.executor(); + } + } + static class DirectResponseChannel implements TransportChannel { final ESLogger logger; final DiscoveryNode localNode; diff --git a/core/src/main/java/org/elasticsearch/transport/local/LocalTransport.java b/core/src/main/java/org/elasticsearch/transport/local/LocalTransport.java index ba067fdabd..d3957245bc 100644 --- a/core/src/main/java/org/elasticsearch/transport/local/LocalTransport.java +++ b/core/src/main/java/org/elasticsearch/transport/local/LocalTransport.java @@ -36,6 +36,7 @@ import org.elasticsearch.common.transport.LocalTransportAddress; import org.elasticsearch.common.transport.TransportAddress; import org.elasticsearch.common.util.concurrent.AbstractRunnable; import org.elasticsearch.common.util.concurrent.EsExecutors; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.ActionNotFoundTransportException; import org.elasticsearch.transport.ConnectTransportException; @@ -72,7 +73,7 @@ import static org.elasticsearch.common.util.concurrent.ConcurrentCollections.new public class LocalTransport extends AbstractLifecycleComponent<Transport> implements Transport { public static final String LOCAL_TRANSPORT_THREAD_NAME_PREFIX = "local_transport"; - private final ThreadPool threadPool; + final ThreadPool threadPool; private final ThreadPoolExecutor workers; private final Version version; private volatile TransportServiceAdapter transportServiceAdapter; @@ -96,7 +97,7 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem int queueSize = this.settings.getAsInt(TRANSPORT_LOCAL_QUEUE, -1); logger.debug("creating [{}] workers, queue_size [{}]", workerCount, queueSize); final ThreadFactory threadFactory = EsExecutors.daemonThreadFactory(this.settings, LOCAL_TRANSPORT_THREAD_NAME_PREFIX); - this.workers = EsExecutors.newFixed(LOCAL_TRANSPORT_THREAD_NAME_PREFIX, workerCount, queueSize, threadFactory); + this.workers = EsExecutors.newFixed(LOCAL_TRANSPORT_THREAD_NAME_PREFIX, workerCount, queueSize, threadFactory, threadPool.getThreadContext()); this.namedWriteableRegistry = namedWriteableRegistry; } @@ -209,6 +210,7 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem status = TransportStatus.setRequest(status); stream.writeByte(status); // 0 for request, 1 for response. + threadPool.getThreadContext().writeTo(stream); stream.writeString(action); request.writeTo(stream); @@ -220,12 +222,11 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem } final byte[] data = stream.bytes().toBytes(); - transportServiceAdapter.sent(data.length); transportServiceAdapter.onRequestSent(node, requestId, action, request, options); - targetTransport.workers().execute(new Runnable() { - @Override - public void run() { + targetTransport.workers().execute(() -> { + ThreadContext threadContext = threadPool.getThreadContext(); + try (ThreadContext.StoredContext context = threadContext.stashContext()) { targetTransport.messageReceived(data, action, LocalTransport.this, version, requestId); } }); @@ -246,8 +247,9 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem long requestId = stream.readLong(); byte status = stream.readByte(); boolean isRequest = TransportStatus.isRequest(status); - if (isRequest) { + ThreadContext threadContext = threadPool.getThreadContext(); + threadContext.readHeaders(stream); handleRequest(stream, requestId, sourceTransport, version); } else { final TransportResponseHandler handler = transportServiceAdapter.onResponseReceived(requestId); @@ -322,6 +324,7 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem logger.warn("Failed to send error message back to client for action [" + action + "]", e); logger.warn("Actual Exception", e1); } + } } @@ -338,15 +341,11 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem } protected void handleParsedResponse(final TransportResponse response, final TransportResponseHandler handler) { - threadPool.executor(handler.executor()).execute(new Runnable() { - @SuppressWarnings({"unchecked"}) - @Override - public void run() { - try { - handler.handleResponse(response); - } catch (Throwable e) { - handleException(handler, new ResponseHandlerFailureTransportException(e)); - } + threadPool.executor(handler.executor()).execute(() -> { + try { + handler.handleResponse(response); + } catch (Throwable e) { + handleException(handler, new ResponseHandlerFailureTransportException(e)); } }); } diff --git a/core/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java b/core/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java index e1e85e9a12..aad31fd8cc 100644 --- a/core/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java +++ b/core/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java @@ -21,6 +21,7 @@ package org.elasticsearch.transport.local; import org.elasticsearch.Version; import org.elasticsearch.common.io.stream.BytesStreamOutput; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.transport.RemoteTransportException; import org.elasticsearch.transport.TransportChannel; import org.elasticsearch.transport.TransportResponse; @@ -79,9 +80,9 @@ public class LocalTransportChannel implements TransportChannel { stream.writeByte(status); // 0 for request, 1 for response. response.writeTo(stream); final byte[] data = stream.bytes().toBytes(); - targetTransport.workers().execute(new Runnable() { - @Override - public void run() { + targetTransport.workers().execute(() -> { + ThreadContext threadContext = targetTransport.threadPool.getThreadContext(); + try (ThreadContext.StoredContext ignore = threadContext.stashContext()){ targetTransport.messageReceived(data, action, sourceTransport, version, null); } }); @@ -97,9 +98,9 @@ public class LocalTransportChannel implements TransportChannel { stream.writeThrowable(tx); final byte[] data = stream.bytes().toBytes(); - targetTransport.workers().execute(new Runnable() { - @Override - public void run() { + targetTransport.workers().execute(() -> { + ThreadContext threadContext = targetTransport.threadPool.getThreadContext(); + try (ThreadContext.StoredContext ignore = threadContext.stashContext()){ targetTransport.messageReceived(data, action, sourceTransport, version, null); } }); diff --git a/core/src/main/java/org/elasticsearch/transport/netty/MessageChannelHandler.java b/core/src/main/java/org/elasticsearch/transport/netty/MessageChannelHandler.java index 8df17f7323..6732b26ddb 100644 --- a/core/src/main/java/org/elasticsearch/transport/netty/MessageChannelHandler.java +++ b/core/src/main/java/org/elasticsearch/transport/netty/MessageChannelHandler.java @@ -30,6 +30,7 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.logging.ESLogger; import org.elasticsearch.common.transport.InetSocketTransportAddress; import org.elasticsearch.common.util.concurrent.AbstractRunnable; +import org.elasticsearch.common.util.concurrent.ThreadContext; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.ActionNotFoundTransportException; import org.elasticsearch.transport.RemoteTransportException; @@ -64,9 +65,11 @@ public class MessageChannelHandler extends SimpleChannelUpstreamHandler { protected final TransportServiceAdapter transportServiceAdapter; protected final NettyTransport transport; protected final String profileName; + private final ThreadContext threadContext; public MessageChannelHandler(NettyTransport transport, ESLogger logger, String profileName) { this.threadPool = transport.threadPool(); + this.threadContext = threadPool.getThreadContext(); this.transportServiceAdapter = transport.transportServiceAdapter(); this.transport = transport; this.logger = logger; @@ -101,7 +104,7 @@ public class MessageChannelHandler extends SimpleChannelUpstreamHandler { // buffer, or in the cumlation buffer, which is cleaned each time StreamInput streamIn = ChannelBufferStreamInputFactory.create(buffer, size); boolean success = false; - try { + try (ThreadContext.StoredContext tCtx = threadContext.stashContext()) { long requestId = streamIn.readLong(); byte status = streamIn.readByte(); Version version = Version.fromId(streamIn.readInt()); @@ -123,8 +126,8 @@ public class MessageChannelHandler extends SimpleChannelUpstreamHandler { streamIn = compressor.streamInput(streamIn); } streamIn.setVersion(version); - if (TransportStatus.isRequest(status)) { + threadContext.readHeaders(streamIn); String action = handleRequest(ctx.getChannel(), streamIn, requestId, version); // Chek the entire message has been read diff --git a/core/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java b/core/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java index 6a6a6c3801..856e45cb63 100644 --- a/core/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java +++ b/core/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java @@ -845,6 +845,7 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem Version version = Version.smallest(this.version, node.version()); stream.setVersion(version); + threadPool.getThreadContext().writeTo(stream); stream.writeString(action); ReleasablePagedBytesReference bytes; |