diff --git a/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/direct/GetDatabaseConfigurationAction.java b/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/direct/GetDatabaseConfigurationAction.java index 68b3ce279a89d..93513e1cc9e6c 100644 --- a/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/direct/GetDatabaseConfigurationAction.java +++ b/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/direct/GetDatabaseConfigurationAction.java @@ -18,7 +18,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xcontent.ToXContentObject; import org.elasticsearch.xcontent.XContentBuilder; @@ -187,7 +187,7 @@ private synchronized Map equalsHashCodeNodesMap() { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { private final String[] databaseIds; diff --git a/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/stats/GeoIpStatsAction.java b/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/stats/GeoIpStatsAction.java index 0fe7cdb4b0d1e..b7792634d6e91 100644 --- a/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/stats/GeoIpStatsAction.java +++ b/modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/stats/GeoIpStatsAction.java @@ -21,7 +21,7 @@ import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; import org.elasticsearch.core.TimeValue; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xcontent.ToXContentObject; import org.elasticsearch.xcontent.XContentBuilder; @@ -68,7 +68,7 @@ public boolean equals(Object obj) { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { public NodeRequest(StreamInput in) throws IOException { super(in); } diff --git a/plugins/examples/security-authorization-engine/src/test/java/org/elasticsearch/example/CustomAuthorizationEngineTests.java b/plugins/examples/security-authorization-engine/src/test/java/org/elasticsearch/example/CustomAuthorizationEngineTests.java index ea3517f315646..262ca55ac8eae 100644 --- a/plugins/examples/security-authorization-engine/src/test/java/org/elasticsearch/example/CustomAuthorizationEngineTests.java +++ b/plugins/examples/security-authorization-engine/src/test/java/org/elasticsearch/example/CustomAuthorizationEngineTests.java @@ -17,6 +17,7 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.index.IndexVersion; import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.xpack.core.security.authc.Authentication; import org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef; @@ -47,7 +48,7 @@ public void testGetAuthorizationInfo() { public void testAuthorizeRunAs() { final String action = "cluster:monitor/foo"; - final TransportRequest request = new TransportRequest() { + final TransportRequest request = new AbstractTransportRequest() { }; CustomAuthorizationEngine engine = new CustomAuthorizationEngine(); // unauthorized @@ -181,7 +182,7 @@ public void testAuthorizeIndexAction() { private RequestInfo getRequestInfo() { final String action = "cluster:monitor/foo"; - final TransportRequest request = new TransportRequest() { + final TransportRequest request = new AbstractTransportRequest() { }; final Authentication authentication = Authentication.newRealmAuthentication( new User("joe", "custom_superuser"), diff --git a/server/src/main/java/org/elasticsearch/action/ActionRequest.java b/server/src/main/java/org/elasticsearch/action/ActionRequest.java index cc2963f49b07e..1fb09062c39d2 100644 --- a/server/src/main/java/org/elasticsearch/action/ActionRequest.java +++ b/server/src/main/java/org/elasticsearch/action/ActionRequest.java @@ -11,11 +11,11 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; -public abstract class ActionRequest extends TransportRequest { +public abstract class ActionRequest extends AbstractTransportRequest { public ActionRequest() { super(); diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/capabilities/TransportNodesCapabilitiesAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/capabilities/TransportNodesCapabilitiesAction.java index e9914ef72a82b..ff403acc445ec 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/capabilities/TransportNodesCapabilitiesAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/capabilities/TransportNodesCapabilitiesAction.java @@ -25,7 +25,7 @@ import org.elasticsearch.rest.RestRequest; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -110,7 +110,7 @@ protected NodeCapability nodeOperation(NodeCapabilitiesRequest request, Task tas return new NodeCapability(supported, transportService.getLocalNode()); } - public static class NodeCapabilitiesRequest extends TransportRequest { + public static class NodeCapabilitiesRequest extends AbstractTransportRequest { private final RestRequest.Method method; private final String path; private final Set parameters; diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/features/TransportNodesFeaturesAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/features/TransportNodesFeaturesAction.java index d20eee96809e8..bd928561caee2 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/features/TransportNodesFeaturesAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/features/TransportNodesFeaturesAction.java @@ -21,13 +21,14 @@ import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; import java.util.List; -@UpdateForV10(owner = UpdateForV10.Owner.CORE_INFRA) // this can be removed in v10. It may be called by v8 nodes to v9 nodes. +@UpdateForV10(owner = UpdateForV10.Owner.CORE_INFRA) +// this can be removed in v10. It may be called by v8 nodes to v9 nodes. public class TransportNodesFeaturesAction extends TransportNodesAction< NodesFeaturesRequest, NodesFeaturesResponse, @@ -82,7 +83,7 @@ protected NodeFeatures nodeOperation(NodeFeaturesRequest request, Task task) { return new NodeFeatures(featureService.getNodeFeatures().keySet(), transportService.getLocalNode()); } - public static class NodeFeaturesRequest extends TransportRequest { + public static class NodeFeaturesRequest extends AbstractTransportRequest { public NodeFeaturesRequest(StreamInput in) throws IOException { super(in); } diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java index 0be9731af4dc7..228dd76f1e2ac 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java @@ -25,8 +25,8 @@ import org.elasticsearch.monitor.jvm.HotThreads; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.LeakTracker; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -106,7 +106,7 @@ protected NodeHotThreads nodeOperation(NodeRequest request, Task task) { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { final HotThreads.RequestOptions requestOptions; diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java index de4f5c9ce7b9f..17cd4525b3916 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/info/TransportNodesInfoAction.java @@ -21,7 +21,7 @@ import org.elasticsearch.node.NodeService; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -95,7 +95,7 @@ protected NodeInfo nodeOperation(NodeInfoRequest nodeRequest, Task task) { ); } - public static class NodeInfoRequest extends TransportRequest { + public static class NodeInfoRequest extends AbstractTransportRequest { private final NodesInfoMetrics nodesInfoMetrics; diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/reload/NodesReloadSecureSettingsRequest.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/reload/NodesReloadSecureSettingsRequest.java index 801047ad2e0cf..26f1a00784394 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/reload/NodesReloadSecureSettingsRequest.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/reload/NodesReloadSecureSettingsRequest.java @@ -24,8 +24,8 @@ import org.elasticsearch.core.RefCounted; import org.elasticsearch.core.Releasables; import org.elasticsearch.tasks.TaskId; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.LeakTracker; -import org.elasticsearch.transport.TransportRequest; import java.io.IOException; import java.util.Arrays; @@ -82,7 +82,7 @@ NodeRequest newNodeRequest() { return new NodeRequest(secureSettingsPassword, refs); } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { @Nullable private final SecureString secureSettingsPassword; diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/shutdown/NodePrevalidateShardPathRequest.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/shutdown/NodePrevalidateShardPathRequest.java index 467b99b4b7955..a70115b8324b5 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/shutdown/NodePrevalidateShardPathRequest.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/shutdown/NodePrevalidateShardPathRequest.java @@ -12,7 +12,7 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.index.shard.ShardId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.Collection; @@ -21,8 +21,8 @@ /** * A node-specific request derived from the corresponding {@link PrevalidateShardPathRequest}. -*/ -public class NodePrevalidateShardPathRequest extends TransportRequest { + */ +public class NodePrevalidateShardPathRequest extends AbstractTransportRequest { private final Set shardIds; diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java index 83769430d4142..66c985d98b01a 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java @@ -34,7 +34,7 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.transport.Transports; @@ -179,7 +179,7 @@ protected NodeStats nodeOperation(NodeStatsRequest request, Task task) { ); } - public static class NodeStatsRequest extends TransportRequest { + public static class NodeStatsRequest extends AbstractTransportRequest { private final NodesStatsRequestParameters nodesStatsRequestParameters; diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/TransportNodesUsageAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/TransportNodesUsageAction.java index 0dc8100cb1785..e96045e8e965a 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/TransportNodesUsageAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/node/usage/TransportNodesUsageAction.java @@ -21,7 +21,7 @@ import org.elasticsearch.search.aggregations.support.AggregationUsageService; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.usage.UsageService; @@ -85,7 +85,7 @@ protected NodeUsage nodeOperation(NodeUsageRequest request, Task task) { return new NodeUsage(clusterService.localNode(), System.currentTimeMillis(), sinceTime, restUsage, aggsUsage); } - public static class NodeUsageRequest extends TransportRequest { + public static class NodeUsageRequest extends AbstractTransportRequest { final boolean restActions; final boolean aggregations; diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java index 64261c75f0a04..0c0c6e41be8ba 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java @@ -29,7 +29,7 @@ import org.elasticsearch.snapshots.SnapshotShardsService; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -148,7 +148,7 @@ protected void writeNodesTo(StreamOutput out, List nodes) th } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { private final List snapshots; @@ -157,7 +157,7 @@ public NodeRequest(StreamInput in) throws IOException { snapshots = in.readCollectionAsList(Snapshot::new); } - NodeRequest(TransportNodesSnapshotsStatus.Request request) { + NodeRequest(Request request) { snapshots = Arrays.asList(request.snapshots); } diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java index 4cb75ae83f187..05a916a076b95 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java @@ -57,10 +57,10 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.RemoteClusterConnection; import org.elasticsearch.transport.RemoteClusterService; import org.elasticsearch.transport.RemoteConnectionInfo; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.transport.Transports; import org.elasticsearch.usage.SearchUsageHolder; @@ -319,7 +319,7 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq ); } - public static class ClusterStatsNodeRequest extends TransportRequest { + public static class ClusterStatsNodeRequest extends AbstractTransportRequest { ClusterStatsNodeRequest() {} diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/find/NodeFindDanglingIndexRequest.java b/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/find/NodeFindDanglingIndexRequest.java index bc1e172f77ce0..21fece55a1f27 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/find/NodeFindDanglingIndexRequest.java +++ b/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/find/NodeFindDanglingIndexRequest.java @@ -11,14 +11,14 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; /** * Used when querying every node in the cluster for a specific dangling index. */ -public class NodeFindDanglingIndexRequest extends TransportRequest { +public class NodeFindDanglingIndexRequest extends AbstractTransportRequest { private final String indexUUID; public NodeFindDanglingIndexRequest(String indexUUID) { diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/list/NodeListDanglingIndicesRequest.java b/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/list/NodeListDanglingIndicesRequest.java index 6baac2dbc8174..cd2e215bdd062 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/list/NodeListDanglingIndicesRequest.java +++ b/server/src/main/java/org/elasticsearch/action/admin/indices/dangling/list/NodeListDanglingIndicesRequest.java @@ -11,14 +11,14 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; /** * Used when querying every node in the cluster for dangling indices, in response to a list request. */ -public class NodeListDanglingIndicesRequest extends TransportRequest { +public class NodeListDanglingIndicesRequest extends AbstractTransportRequest { /** * Filter the response by index UUID. Leave as null to find all indices. */ diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java b/server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java index 5d184f56dd748..c0a3e568ffeeb 100644 --- a/server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java +++ b/server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java @@ -26,8 +26,8 @@ import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportChannel; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportService; @@ -100,7 +100,7 @@ protected void shardOperationOnReplica(ShardFlushRequest request, IndexShard rep // TODO: Remove this transition in 9.0 private static final String PRE_SYNCED_FLUSH_ACTION_NAME = "internal:indices/flush/synced/pre"; - private static class PreShardSyncedFlushRequest extends TransportRequest { + private static class PreShardSyncedFlushRequest extends AbstractTransportRequest { private final ShardId shardId; private PreShardSyncedFlushRequest(StreamInput in) throws IOException { diff --git a/server/src/main/java/org/elasticsearch/action/search/CanMatchNodeRequest.java b/server/src/main/java/org/elasticsearch/action/search/CanMatchNodeRequest.java index 17c272d75d014..93a9e4d87f29f 100644 --- a/server/src/main/java/org/elasticsearch/action/search/CanMatchNodeRequest.java +++ b/server/src/main/java/org/elasticsearch/action/search/CanMatchNodeRequest.java @@ -29,7 +29,7 @@ import org.elasticsearch.search.internal.ShardSearchRequest; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.ArrayList; @@ -41,7 +41,7 @@ /** * Node-level request used during can-match phase */ -public class CanMatchNodeRequest extends TransportRequest implements IndicesRequest { +public class CanMatchNodeRequest extends AbstractTransportRequest implements IndicesRequest { private final SearchSourceBuilder source; private final List shards; diff --git a/server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java b/server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java index 545e28f64749d..0db16c2960dd7 100644 --- a/server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java +++ b/server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java @@ -50,13 +50,13 @@ import org.elasticsearch.tasks.TaskCancelledException; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.LeakTracker; import org.elasticsearch.transport.SendRequestTransportException; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportActionProxy; import org.elasticsearch.transport.TransportChannel; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportResponse; import org.elasticsearch.transport.TransportResponseHandler; @@ -286,7 +286,7 @@ public boolean decRef() { /** * Request for starting the query phase for multiple shards. */ - public static final class NodeQueryRequest extends TransportRequest implements IndicesRequest { + public static final class NodeQueryRequest extends AbstractTransportRequest implements IndicesRequest { private final List shards; private final SearchRequest searchRequest; private final Map aliasFilters; @@ -368,7 +368,7 @@ public void writeTo(StreamOutput out) throws IOException { * Check if, based on already collected results, a shard search can be updated with a lower search threshold than is current set. * When the query executes via batched execution, data nodes this take into account the results of queries run against shards local * to the datanode. On the coordinating node results received from all data nodes are taken into account. - * + *

* See {@link BottomSortValuesCollector} for details. */ private static ShardSearchRequest tryRewriteWithUpdatedSortValue( diff --git a/server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java b/server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java index fc77daafac12a..1e0fa28889c97 100644 --- a/server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java +++ b/server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java @@ -48,11 +48,11 @@ import org.elasticsearch.search.rank.feature.RankFeatureShardRequest; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.RemoteClusterService; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportActionProxy; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportResponse; @@ -67,7 +67,7 @@ import java.util.function.BiFunction; /** - * An encapsulation of {@link org.elasticsearch.search.SearchService} operations exposed through + * An encapsulation of {@link SearchService} operations exposed through * transport. */ public class SearchTransportService { @@ -323,7 +323,7 @@ public Map getPendingSearchRequests() { return new HashMap<>(clientConnections); } - static class ScrollFreeContextRequest extends TransportRequest { + static class ScrollFreeContextRequest extends AbstractTransportRequest { private final ShardSearchContextId contextId; ScrollFreeContextRequest(ShardSearchContextId contextId) { @@ -347,7 +347,7 @@ public ShardSearchContextId id() { } - private static class ClearScrollContextsRequest extends TransportRequest { + private static class ClearScrollContextsRequest extends AbstractTransportRequest { ClearScrollContextsRequest() {} ClearScrollContextsRequest(StreamInput in) throws IOException { diff --git a/server/src/main/java/org/elasticsearch/action/search/TransportOpenPointInTimeAction.java b/server/src/main/java/org/elasticsearch/action/search/TransportOpenPointInTimeAction.java index 3ae80d44c8e21..357c65f2a54c6 100644 --- a/server/src/main/java/org/elasticsearch/action/search/TransportOpenPointInTimeAction.java +++ b/server/src/main/java/org/elasticsearch/action/search/TransportOpenPointInTimeAction.java @@ -39,10 +39,10 @@ import org.elasticsearch.search.internal.ShardSearchContextId; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportActionProxy; import org.elasticsearch.transport.TransportChannel; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportResponseHandler; import org.elasticsearch.transport.TransportService; @@ -92,12 +92,7 @@ public TransportOpenPointInTimeAction( ShardOpenReaderRequest::new, new ShardOpenReaderRequestHandler() ); - TransportActionProxy.registerProxyAction( - transportService, - OPEN_SHARD_READER_CONTEXT_NAME, - false, - TransportOpenPointInTimeAction.ShardOpenReaderResponse::new - ); + TransportActionProxy.registerProxyAction(transportService, OPEN_SHARD_READER_CONTEXT_NAME, false, ShardOpenReaderResponse::new); } @Override @@ -278,7 +273,7 @@ boolean buildPointInTimeFromSearchResults() { } } - private static final class ShardOpenReaderRequest extends TransportRequest implements IndicesRequest { + private static final class ShardOpenReaderRequest extends AbstractTransportRequest implements IndicesRequest { final ShardId shardId; final OriginalIndices originalIndices; final TimeValue keepAlive; diff --git a/server/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java b/server/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java index 5ad20ebdd5d1a..a6f5d6e6e6e8c 100644 --- a/server/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java +++ b/server/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardRequest.java @@ -15,11 +15,11 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.index.shard.ShardId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; -public abstract class BroadcastShardRequest extends TransportRequest implements IndicesRequest { +public abstract class BroadcastShardRequest extends AbstractTransportRequest implements IndicesRequest { private final ShardId shardId; diff --git a/server/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java b/server/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java index cd3ffaf7bf431..899f83f5f6fb1 100644 --- a/server/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java +++ b/server/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java @@ -42,8 +42,8 @@ import org.elasticsearch.core.FixForMultiProject; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportChannel; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportResponse; @@ -222,7 +222,7 @@ protected abstract void shardOperation( * Resolves a list of concrete index names. Override this if index names should be resolved differently than normal. * * @param clusterState the cluster state - * @param request the underlying request + * @param request the underlying request * @return a list of concrete index names that this action should operate on */ protected String[] resolveConcreteIndexNames(ClusterState clusterState, Request request) { @@ -465,7 +465,7 @@ public String toString() { }.run(task, shards.iterator(), listener); } - class NodeRequest extends TransportRequest implements IndicesRequest { + class NodeRequest extends AbstractTransportRequest implements IndicesRequest { private final Request indicesLevelRequest; private final List shards; private final String nodeId; diff --git a/server/src/main/java/org/elasticsearch/action/support/master/TermOverridingMasterNodeRequest.java b/server/src/main/java/org/elasticsearch/action/support/master/TermOverridingMasterNodeRequest.java index be3e749d058c9..c81655ff2f02f 100644 --- a/server/src/main/java/org/elasticsearch/action/support/master/TermOverridingMasterNodeRequest.java +++ b/server/src/main/java/org/elasticsearch/action/support/master/TermOverridingMasterNodeRequest.java @@ -9,13 +9,14 @@ package org.elasticsearch.action.support.master; +import org.elasticsearch.action.IndicesRequest; import org.elasticsearch.common.Strings; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.logging.LogManager; import org.elasticsearch.logging.Logger; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.net.InetSocketAddress; @@ -26,13 +27,13 @@ * MasterNodeRequest#masterTerm()} sent out over the wire. *

* Note that in production this is only used for sending the request out, so there's no need to preserve other marker interfaces such - * as {@link org.elasticsearch.action.IndicesRequest} or {@link org.elasticsearch.action.IndicesRequest.Replaceable} on the wrapped request. + * as {@link IndicesRequest} or {@link IndicesRequest.Replaceable} on the wrapped request. * The receiving node will deserialize a request without a wrapper, with the correct interfaces and the appropriate master term stored * directly in {@link MasterNodeRequest#masterTerm()}. However in tests sometimes we want to intercept the request as it's being sent, for * which it may be necessary to use the test utility {@code MasterNodeRequestHelper#unwrapTermOverride} to remove the wrapper and access the * inner request. */ -class TermOverridingMasterNodeRequest extends TransportRequest { +class TermOverridingMasterNodeRequest extends AbstractTransportRequest { private static final Logger logger = LogManager.getLogger(TermOverridingMasterNodeRequest.class); diff --git a/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java b/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java index ba17b193b7dc1..27d5327711634 100644 --- a/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java +++ b/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java @@ -63,6 +63,7 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.ConnectTransportException; import org.elasticsearch.transport.RawIndexingDataTransportRequest; import org.elasticsearch.transport.TransportChannel; @@ -84,7 +85,7 @@ /** * Base class for requests that should be executed on a primary copy followed by replica copies. * Subclasses can resolve the target shard and provide implementation for primary and replica operations. - * + *

* The action samples cluster state on the receiving node to reroute to node with primary copy and on the * primary node to validate request before primary operation followed by sampling state again for resolving * nodes with replica copies to perform replication. @@ -793,7 +794,7 @@ private IndexShard getIndexShard(final ShardId shardId) { * Responsible for routing and retrying failed operations on the primary. * The actual primary operation is done in {@link ReplicationOperation} on the * node with primary copy. - * + *

* Resolves index and shard id for the request before routing it to target node */ final class ReroutePhase extends AbstractRunnable { @@ -1331,12 +1332,16 @@ public void markShardCopyAsStaleIfNeeded(ShardId shardId, String allocationId, l } } - /** a wrapper class to encapsulate a request when being sent to a specific allocation id **/ - public static class ConcreteShardRequest extends TransportRequest + /** + * a wrapper class to encapsulate a request when being sent to a specific allocation id + **/ + public static class ConcreteShardRequest extends AbstractTransportRequest implements RawIndexingDataTransportRequest { - /** {@link AllocationId#getId()} of the shard this request is sent to **/ + /** + * {@link AllocationId#getId()} of the shard this request is sent to + **/ private final String targetAllocationID; private final long primaryTerm; private final R request; @@ -1346,7 +1351,7 @@ public static class ConcreteShardRequest extends Tra // is only true if sentFromLocalReroute is true. private final boolean localRerouteInitiatedByNodeClient; - public ConcreteShardRequest(Writeable.Reader requestReader, StreamInput in) throws IOException { + public ConcreteShardRequest(Reader requestReader, StreamInput in) throws IOException { targetAllocationID = in.readString(); primaryTerm = in.readVLong(); sentFromLocalReroute = false; @@ -1460,7 +1465,7 @@ protected static final class ConcreteReplicaRequest private final long globalCheckpoint; private final long maxSeqNoOfUpdatesOrDeletes; - public ConcreteReplicaRequest(Writeable.Reader requestReader, StreamInput in) throws IOException { + public ConcreteReplicaRequest(Reader requestReader, StreamInput in) throws IOException { super(requestReader, in); globalCheckpoint = in.readZLong(); maxSeqNoOfUpdatesOrDeletes = in.readZLong(); diff --git a/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java b/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java index 46417839a1458..1c93f53869aa8 100644 --- a/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java +++ b/server/src/main/java/org/elasticsearch/action/support/tasks/TransportTasksAction.java @@ -30,8 +30,8 @@ import org.elasticsearch.tasks.CancellableTask; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportChannel; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportResponse; @@ -243,10 +243,11 @@ protected abstract TasksResponse newResponse( /** * Perform the required operation on the task. It is OK start an asynchronous operation or to throw an exception but not both. + * * @param actionTask The related transport action task. Can be used to create a task ID to handle upstream transport cancellations. - * @param request the original transport request - * @param task the task on which the operation is taking place - * @param listener the listener to signal. + * @param request the original transport request + * @param task the task on which the operation is taking place + * @param listener the listener to signal. */ protected abstract void taskOperation( CancellableTask actionTask, @@ -271,7 +272,7 @@ public void messageReceived(final NodeTaskRequest request, final TransportChanne } } - private class NodeTaskRequest extends TransportRequest { + private class NodeTaskRequest extends AbstractTransportRequest { private final TasksRequest tasksRequest; protected NodeTaskRequest(StreamInput in) throws IOException { diff --git a/server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java b/server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java index f6e2a3283898c..c2820e6da630e 100644 --- a/server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java +++ b/server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java @@ -49,10 +49,12 @@ import org.elasticsearch.index.shard.IndexLongFieldRange; import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.index.shard.ShardLongFieldRange; +import org.elasticsearch.indices.cluster.IndicesClusterStateService; import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.node.NodeClosedException; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.ConnectTransportException; import org.elasticsearch.transport.RemoteTransportException; import org.elasticsearch.transport.TransportChannel; @@ -172,13 +174,13 @@ private static boolean isMasterChannelException(Throwable exp) { * that the shard should be failed because a write made it into the primary but was not replicated to this shard copy. If the shard * does not exist anymore but still has an entry in the in-sync set, remove its allocation id from the in-sync set. * - * @param shardId shard id of the shard to fail - * @param allocationId allocation id of the shard to fail - * @param primaryTerm the primary term associated with the primary shard that is failing the shard. Must be strictly positive. - * @param markAsStale whether or not to mark a failing shard as stale (eg. removing from in-sync set) when failing the shard. - * @param message the reason for the failure - * @param failure the underlying cause of the failure - * @param listener callback upon completion of the request + * @param shardId shard id of the shard to fail + * @param allocationId allocation id of the shard to fail + * @param primaryTerm the primary term associated with the primary shard that is failing the shard. Must be strictly positive. + * @param markAsStale whether or not to mark a failing shard as stale (eg. removing from in-sync set) when failing the shard. + * @param message the reason for the failure + * @param failure the underlying cause of the failure + * @param listener callback upon completion of the request */ public void remoteShardFailed( final ShardId shardId, @@ -203,7 +205,7 @@ int remoteShardRequestsInFlight() { /** * Clears out {@link #remoteShardStateUpdateDeduplicator}. Called by - * {@link org.elasticsearch.indices.cluster.IndicesClusterStateService} in case of a master failover to enable sending fresh requests + * {@link IndicesClusterStateService} in case of a master failover to enable sending fresh requests * to the new master right away on master failover. * This method is best effort in so far that it might clear out valid requests in edge cases during master failover. This is not an * issue functionally and merely results in some unnecessary transport requests. @@ -308,7 +310,7 @@ public ShardFailedClusterStateTaskExecutor(AllocationService allocationService, @Override public ClusterState execute(BatchExecutionContext batchExecutionContext) throws Exception { - List> tasksToBeApplied = new ArrayList<>(); + List> tasksToBeApplied = new ArrayList<>(); List failedShardsToBeApplied = new ArrayList<>(); List staleShardsToBeApplied = new ArrayList<>(); final ClusterState initialState = batchExecutionContext.initialState(); @@ -443,7 +445,7 @@ public void clusterStatePublished(ClusterState newClusterState) { } } - public static class FailedShardEntry extends TransportRequest { + public static class FailedShardEntry extends AbstractTransportRequest { final ShardId shardId; final String allocationId; final long primaryTerm; @@ -611,7 +613,8 @@ public void messageReceived(StartedShardEntry request, TransportChannel channel, /** * Holder of the pair of time ranges needed in cluster state - one for @timestamp, the other for 'event.ingested'. * Since 'event.ingested' was added well after @timestamp, it can be UNKNOWN when @timestamp range is present. - * @param timestampRange range for @timestamp + * + * @param timestampRange range for @timestamp * @param eventIngestedRange range for event.ingested */ record ClusterStateTimeRanges(IndexLongFieldRange timestampRange, IndexLongFieldRange eventIngestedRange) {} @@ -833,7 +836,7 @@ public void clusterStatePublished(ClusterState newClusterState) { } } - public static class StartedShardEntry extends TransportRequest { + public static class StartedShardEntry extends AbstractTransportRequest { final ShardId shardId; final String allocationId; final long primaryTerm; diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java b/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java index f3b4da611811a..37ed3a7e739fd 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java @@ -30,12 +30,12 @@ import org.elasticsearch.monitor.NodeHealthService; import org.elasticsearch.monitor.StatusInfo; import org.elasticsearch.threadpool.ThreadPool.Names; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.ConnectTransportException; import org.elasticsearch.transport.ReceiveTimeoutTransportException; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportConnectionListener; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportRequestOptions.Type; import org.elasticsearch.transport.TransportResponseHandler; @@ -434,7 +434,7 @@ public String toString() { } } - public static class FollowerCheckRequest extends TransportRequest { + public static class FollowerCheckRequest extends AbstractTransportRequest { private final long term; diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java index ec87df9da2d5e..8f302c2e2ee8c 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java @@ -42,9 +42,9 @@ import org.elasticsearch.monitor.StatusInfo; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.threadpool.ThreadPool.Names; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.ConnectTransportException; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportResponseHandler; import org.elasticsearch.transport.TransportService; @@ -200,7 +200,7 @@ private void unregisterAndReleaseConnection(DiscoveryNode destination, Releasabl /** * Saves information about a join failure. The failure information may be logged later via either {@link FailedJoinAttempt#logNow} * or {@link FailedJoinAttempt#lastFailedJoinAttempt}. - * + *

* Package-private for testing. */ static class FailedJoinAttempt { @@ -212,7 +212,7 @@ static class FailedJoinAttempt { /** * @param destination the master node targeted by the join request. * @param joinRequest the join request that was sent to the perceived master node. - * @param exception the error response received in reply to the join request attempt. + * @param exception the error response received in reply to the join request attempt. */ FailedJoinAttempt(DiscoveryNode destination, JoinRequest joinRequest, ElasticsearchException exception) { this.destination = destination; @@ -610,7 +610,7 @@ private static class PendingJoinInfo { static final String PENDING_JOIN_CONNECT_FAILED = "failed to connect"; static final String PENDING_JOIN_FAILED = "failed"; - static class JoinPingRequest extends TransportRequest { + static class JoinPingRequest extends AbstractTransportRequest { JoinPingRequest() {} JoinPingRequest(StreamInput in) throws IOException { diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinRequest.java b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinRequest.java index 5f499634fdbaa..879a165dda47e 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinRequest.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinRequest.java @@ -12,14 +12,14 @@ import org.elasticsearch.cluster.version.CompatibilityVersions; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.Objects; import java.util.Optional; import java.util.Set; -public class JoinRequest extends TransportRequest { +public class JoinRequest extends AbstractTransportRequest { /** * The sending (i.e. joining) node. diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java b/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java index 9cc079e4d897d..bbf0884cb4c26 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java @@ -28,13 +28,13 @@ import org.elasticsearch.monitor.NodeHealthService; import org.elasticsearch.monitor.StatusInfo; import org.elasticsearch.threadpool.ThreadPool.Names; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.ConnectTransportException; import org.elasticsearch.transport.NodeDisconnectedException; import org.elasticsearch.transport.ReceiveTimeoutTransportException; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportConnectionListener; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportRequestOptions.Type; import org.elasticsearch.transport.TransportResponseHandler; @@ -397,7 +397,7 @@ public String toString() { } } - static class LeaderCheckRequest extends TransportRequest { + static class LeaderCheckRequest extends AbstractTransportRequest { private final DiscoveryNode sender; diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/PreVoteRequest.java b/server/src/main/java/org/elasticsearch/cluster/coordination/PreVoteRequest.java index 630bf4abe18bd..a0c73cb8cf928 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/PreVoteRequest.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/PreVoteRequest.java @@ -12,12 +12,12 @@ import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.Objects; -public class PreVoteRequest extends TransportRequest { +public class PreVoteRequest extends AbstractTransportRequest { private final DiscoveryNode sourceNode; private final long currentTerm; diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/StartJoinRequest.java b/server/src/main/java/org/elasticsearch/cluster/coordination/StartJoinRequest.java index 68e694ca63723..deb1e625536e6 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/StartJoinRequest.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/StartJoinRequest.java @@ -11,19 +11,19 @@ import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; /** * Represents the action of requesting a join vote (see {@link Join}) from a node. - * + *

* A {@link StartJoinRequest} is broadcast to each node in the cluster, requesting * that each node join the new cluster formed around the master candidate node in a * new term. The sender is either the new master candidate or the current master * abdicating to another eligible node in the cluster. */ -public class StartJoinRequest extends TransportRequest { +public class StartJoinRequest extends AbstractTransportRequest { private final DiscoveryNode masterCandidateNode; diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/TermVersionRequest.java b/server/src/main/java/org/elasticsearch/cluster/coordination/TermVersionRequest.java index 878abd66abfa2..88738e102a021 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/TermVersionRequest.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/TermVersionRequest.java @@ -12,11 +12,11 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; -abstract class TermVersionRequest extends TransportRequest implements Writeable { +abstract class TermVersionRequest extends AbstractTransportRequest implements Writeable { protected final DiscoveryNode sourceNode; protected final long term; protected final long version; diff --git a/server/src/main/java/org/elasticsearch/discovery/PeersRequest.java b/server/src/main/java/org/elasticsearch/discovery/PeersRequest.java index afb376718ba34..e2967f1132e77 100644 --- a/server/src/main/java/org/elasticsearch/discovery/PeersRequest.java +++ b/server/src/main/java/org/elasticsearch/discovery/PeersRequest.java @@ -12,13 +12,13 @@ import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.List; import java.util.Objects; -public class PeersRequest extends TransportRequest { +public class PeersRequest extends AbstractTransportRequest { private final DiscoveryNode sourceNode; private final List knownPeers; diff --git a/server/src/main/java/org/elasticsearch/gateway/LocalAllocateDangledIndices.java b/server/src/main/java/org/elasticsearch/gateway/LocalAllocateDangledIndices.java index e34be2d4a1ef2..665811895fd7f 100644 --- a/server/src/main/java/org/elasticsearch/gateway/LocalAllocateDangledIndices.java +++ b/server/src/main/java/org/elasticsearch/gateway/LocalAllocateDangledIndices.java @@ -36,8 +36,8 @@ import org.elasticsearch.index.IndexVersion; import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportChannel; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportResponse; import org.elasticsearch.transport.TransportService; @@ -251,7 +251,7 @@ private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String clusterService.submitUnbatchedStateUpdateTask(source, task); } - public static class AllocateDangledRequest extends TransportRequest { + public static class AllocateDangledRequest extends AbstractTransportRequest { DiscoveryNode fromNode; IndexMetadata[] indices; diff --git a/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java b/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java index d5d652ec05088..87742d0d71fb2 100644 --- a/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java +++ b/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java @@ -38,7 +38,7 @@ import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xcontent.NamedXContentRegistry; @@ -229,7 +229,7 @@ protected void writeNodesTo(StreamOutput out, List nod } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { private final ShardId shardId; @Nullable diff --git a/server/src/main/java/org/elasticsearch/health/stats/HealthApiStatsAction.java b/server/src/main/java/org/elasticsearch/health/stats/HealthApiStatsAction.java index fd3fadcbf1f98..e5768a57df924 100644 --- a/server/src/main/java/org/elasticsearch/health/stats/HealthApiStatsAction.java +++ b/server/src/main/java/org/elasticsearch/health/stats/HealthApiStatsAction.java @@ -21,7 +21,7 @@ import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.metrics.Counters; import org.elasticsearch.core.Nullable; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.List; @@ -50,7 +50,7 @@ public String toString() { return "health_api_stats"; } - public static class Node extends TransportRequest { + public static class Node extends AbstractTransportRequest { public Node(StreamInput in) throws IOException { super(in); diff --git a/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryHandoffPrimaryContextRequest.java b/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryHandoffPrimaryContextRequest.java index ad43edeeddad7..b2fdfd4e15296 100644 --- a/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryHandoffPrimaryContextRequest.java +++ b/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryHandoffPrimaryContextRequest.java @@ -13,14 +13,14 @@ import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.index.seqno.ReplicationTracker; import org.elasticsearch.index.shard.ShardId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; /** * The request object to handoff the primary context to the relocation target. */ -class RecoveryHandoffPrimaryContextRequest extends TransportRequest { +class RecoveryHandoffPrimaryContextRequest extends AbstractTransportRequest { private final long recoveryId; private final ShardId shardId; diff --git a/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTransportRequest.java b/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTransportRequest.java index ba0e1c45fc7de..773083cf1a7e9 100644 --- a/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTransportRequest.java +++ b/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryTransportRequest.java @@ -12,11 +12,11 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.index.shard.ShardId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; -public abstract class RecoveryTransportRequest extends TransportRequest { +public abstract class RecoveryTransportRequest extends AbstractTransportRequest { private final long requestSeqNo; diff --git a/server/src/main/java/org/elasticsearch/indices/recovery/ReestablishRecoveryRequest.java b/server/src/main/java/org/elasticsearch/indices/recovery/ReestablishRecoveryRequest.java index ec13a740ebe91..50390821cecf3 100644 --- a/server/src/main/java/org/elasticsearch/indices/recovery/ReestablishRecoveryRequest.java +++ b/server/src/main/java/org/elasticsearch/indices/recovery/ReestablishRecoveryRequest.java @@ -12,14 +12,14 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.index.shard.ShardId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; /** * Represents a request for starting a peer recovery. */ -public class ReestablishRecoveryRequest extends TransportRequest { +public class ReestablishRecoveryRequest extends AbstractTransportRequest { private final long recoveryId; private final ShardId shardId; diff --git a/server/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java b/server/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java index 2f20c61324850..83c9d9e3f1af2 100644 --- a/server/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java +++ b/server/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java @@ -17,14 +17,14 @@ import org.elasticsearch.index.seqno.SequenceNumbers; import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.index.store.Store; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; /** * Represents a request for starting a peer recovery. */ -public class StartRecoveryRequest extends TransportRequest { +public class StartRecoveryRequest extends AbstractTransportRequest { private final long recoveryId; private final ShardId shardId; diff --git a/server/src/main/java/org/elasticsearch/indices/store/IndicesStore.java b/server/src/main/java/org/elasticsearch/indices/store/IndicesStore.java index 44b8ff4dc2a8f..e1bc3ef7d2f34 100644 --- a/server/src/main/java/org/elasticsearch/indices/store/IndicesStore.java +++ b/server/src/main/java/org/elasticsearch/indices/store/IndicesStore.java @@ -47,9 +47,9 @@ import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportChannel; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportResponse; import org.elasticsearch.transport.TransportResponseHandler; @@ -462,7 +462,7 @@ private IndexShard getShard(ShardActiveRequest request) { } - private static class ShardActiveRequest extends TransportRequest { + private static class ShardActiveRequest extends AbstractTransportRequest { private final TimeValue timeout; private final ClusterName clusterName; private final String indexUUID; diff --git a/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetadata.java b/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetadata.java index 237ec46d936d2..8d3008bc28758 100644 --- a/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetadata.java +++ b/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetadata.java @@ -11,6 +11,7 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import org.apache.lucene.index.IndexNotFoundException; import org.elasticsearch.ElasticsearchException; import org.elasticsearch.TransportVersions; import org.elasticsearch.action.ActionType; @@ -45,7 +46,7 @@ import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -142,7 +143,7 @@ private StoreFilesMetadata listStoreMetadata(NodeRequest request) throws IOExcep ); exists = true; return storeFilesMetadata; - } catch (org.apache.lucene.index.IndexNotFoundException e) { + } catch (IndexNotFoundException e) { logger.trace(() -> "[" + shardId + "] node is missing index, responding with empty", e); return StoreFilesMetadata.EMPTY; } catch (IOException e) { @@ -306,7 +307,7 @@ protected void writeNodesTo(StreamOutput out, List nodes } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { private final ShardId shardId; @Nullable diff --git a/server/src/main/java/org/elasticsearch/monitor/metrics/IndexModeStatsActionType.java b/server/src/main/java/org/elasticsearch/monitor/metrics/IndexModeStatsActionType.java index f64bbb6ec3d50..e6b17b179d1a0 100644 --- a/server/src/main/java/org/elasticsearch/monitor/metrics/IndexModeStatsActionType.java +++ b/server/src/main/java/org/elasticsearch/monitor/metrics/IndexModeStatsActionType.java @@ -26,7 +26,7 @@ import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -88,7 +88,7 @@ public Map stats() { } } - public static final class NodeRequest extends TransportRequest { + public static final class NodeRequest extends AbstractTransportRequest { NodeRequest() { } diff --git a/server/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java b/server/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java index 7126dc62806fd..ffcd74fc2c486 100644 --- a/server/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java +++ b/server/src/main/java/org/elasticsearch/search/fetch/ShardFetchRequest.java @@ -23,7 +23,7 @@ import org.elasticsearch.search.rank.RankDocShardInfo; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.List; @@ -33,7 +33,7 @@ * Shard level fetch base request. Holds all the info needed to execute a fetch. * Used with search scroll as the original request doesn't hold indices. */ -public class ShardFetchRequest extends TransportRequest { +public class ShardFetchRequest extends AbstractTransportRequest { private final ShardSearchContextId contextId; diff --git a/server/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java b/server/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java index 76b553f568ee8..2e709a37f4507 100644 --- a/server/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java +++ b/server/src/main/java/org/elasticsearch/search/internal/InternalScrollSearchRequest.java @@ -16,12 +16,12 @@ import org.elasticsearch.core.TimeValue; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.Map; -public class InternalScrollSearchRequest extends TransportRequest { +public class InternalScrollSearchRequest extends AbstractTransportRequest { private ShardSearchContextId contextId; diff --git a/server/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java b/server/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java index 5e4ffbdba9ad2..be504b77713c4 100644 --- a/server/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java +++ b/server/src/main/java/org/elasticsearch/search/internal/ShardSearchRequest.java @@ -51,7 +51,7 @@ import org.elasticsearch.search.sort.FieldSortBuilder; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.ArrayList; @@ -68,7 +68,7 @@ * It provides all the methods that the {@link SearchContext} needs. * Provides a cache key based on its content that can be used to cache shard level response. */ -public class ShardSearchRequest extends TransportRequest implements IndicesRequest { +public class ShardSearchRequest extends AbstractTransportRequest implements IndicesRequest { private final String clusterAlias; private final ShardId shardId; private final int shardRequestIndex; diff --git a/server/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java b/server/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java index 005dba90d9810..7c0b9ac2c0118 100644 --- a/server/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java +++ b/server/src/main/java/org/elasticsearch/search/query/QuerySearchRequest.java @@ -22,12 +22,12 @@ import org.elasticsearch.search.internal.ShardSearchRequest; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.Map; -public class QuerySearchRequest extends TransportRequest implements IndicesRequest { +public class QuerySearchRequest extends AbstractTransportRequest implements IndicesRequest { private final ShardSearchContextId contextId; private final AggregatedDfs dfs; diff --git a/server/src/main/java/org/elasticsearch/search/rank/feature/RankFeatureShardRequest.java b/server/src/main/java/org/elasticsearch/search/rank/feature/RankFeatureShardRequest.java index 41b271a6beb53..d6c10f15adf80 100644 --- a/server/src/main/java/org/elasticsearch/search/rank/feature/RankFeatureShardRequest.java +++ b/server/src/main/java/org/elasticsearch/search/rank/feature/RankFeatureShardRequest.java @@ -18,7 +18,7 @@ import org.elasticsearch.search.internal.ShardSearchContextId; import org.elasticsearch.search.internal.ShardSearchRequest; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.List; @@ -29,7 +29,7 @@ * Shard level request for extracting all needed feature for a global reranker */ -public class RankFeatureShardRequest extends TransportRequest implements IndicesRequest { +public class RankFeatureShardRequest extends AbstractTransportRequest implements IndicesRequest { private final OriginalIndices originalIndices; private final ShardSearchRequest shardSearchRequest; diff --git a/server/src/main/java/org/elasticsearch/tasks/TaskCancellationService.java b/server/src/main/java/org/elasticsearch/tasks/TaskCancellationService.java index 7155901ca479e..63b1a0d1b1756 100644 --- a/server/src/main/java/org/elasticsearch/tasks/TaskCancellationService.java +++ b/server/src/main/java/org/elasticsearch/tasks/TaskCancellationService.java @@ -27,12 +27,12 @@ import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.util.concurrent.EsExecutors; import org.elasticsearch.common.util.concurrent.ListenableFuture; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.NodeDisconnectedException; import org.elasticsearch.transport.NodeNotConnectedException; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportChannel; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportResponseHandler; @@ -299,7 +299,7 @@ private static boolean isUnimportantBanFailure(Throwable cause) { return cause instanceof NodeDisconnectedException || cause instanceof NodeNotConnectedException; } - private static class BanParentTaskRequest extends TransportRequest { + private static class BanParentTaskRequest extends AbstractTransportRequest { private final TaskId parentTaskId; private final boolean ban; @@ -375,7 +375,7 @@ public void messageReceived(final BanParentTaskRequest request, final TransportC } } - private static class CancelChildRequest extends TransportRequest { + private static class CancelChildRequest extends AbstractTransportRequest { private final TaskId parentTaskId; private final long childRequestId; diff --git a/server/src/main/java/org/elasticsearch/transport/AbstractTransportRequest.java b/server/src/main/java/org/elasticsearch/transport/AbstractTransportRequest.java new file mode 100644 index 0000000000000..e51927a5b63fd --- /dev/null +++ b/server/src/main/java/org/elasticsearch/transport/AbstractTransportRequest.java @@ -0,0 +1,89 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +package org.elasticsearch.transport; + +import org.elasticsearch.common.io.stream.StreamInput; +import org.elasticsearch.common.io.stream.StreamOutput; +import org.elasticsearch.core.Nullable; +import org.elasticsearch.tasks.TaskId; + +import java.io.IOException; +import java.net.InetSocketAddress; + +public abstract class AbstractTransportRequest extends TransportMessage implements TransportRequest { + @Nullable // set by the transport service on inbound messages; unset on outbound messages + private InetSocketAddress remoteAddress; + + /** + * Parent of this request. Defaults to {@link TaskId#EMPTY_TASK_ID}, meaning "no parent". + */ + private TaskId parentTaskId = TaskId.EMPTY_TASK_ID; + + /** + * Request ID. Defaults to -1, meaning "no request ID is set". + */ + private volatile long requestId = -1; + + public AbstractTransportRequest() {} + + public AbstractTransportRequest(StreamInput in) throws IOException { + parentTaskId = TaskId.readFromStream(in); + } + + @Override + public void remoteAddress(InetSocketAddress remoteAddress) { + this.remoteAddress = remoteAddress; + } + + @Nullable // set by the transport service on inbound messages; unset on outbound messages + @Override + public InetSocketAddress remoteAddress() { + return remoteAddress; + } + + /** + * Set a reference to task that created this request. + */ + @Override + public void setParentTask(TaskId taskId) { + this.parentTaskId = taskId; + } + + /** + * Get a reference to the task that created this request. Defaults to {@link TaskId#EMPTY_TASK_ID}, meaning "there is no parent". + */ + @Override + public TaskId getParentTask() { + return parentTaskId; + } + + /** + * Set the request ID of this request. + */ + @Override + public void setRequestId(long requestId) { + this.requestId = requestId; + } + + @Override + public long getRequestId() { + return requestId; + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + parentTaskId.writeTo(out); + } + + @Override + public String toString() { + return getClass().getName() + "/" + getParentTask(); + } +} diff --git a/server/src/main/java/org/elasticsearch/transport/BytesTransportRequest.java b/server/src/main/java/org/elasticsearch/transport/BytesTransportRequest.java index 0db3de9abdb7b..a16411cb25f1d 100644 --- a/server/src/main/java/org/elasticsearch/transport/BytesTransportRequest.java +++ b/server/src/main/java/org/elasticsearch/transport/BytesTransportRequest.java @@ -21,7 +21,7 @@ * A specialized, bytes only request, that can potentially be optimized on the network * layer, specifically for the same large buffer send to several nodes. */ -public class BytesTransportRequest extends TransportRequest { +public class BytesTransportRequest extends AbstractTransportRequest { final ReleasableBytesReference bytes; private final TransportVersion version; diff --git a/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java b/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java index f5b2f68ccbd94..84b5be57a97d3 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java @@ -125,7 +125,7 @@ public boolean decRef() { } } - static class ProxyRequest extends TransportRequest { + static class ProxyRequest extends AbstractTransportRequest { final T wrapped; final DiscoveryNode targetNode; diff --git a/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java b/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java index d0e456f1326a4..9ba910d8bac0b 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java @@ -337,7 +337,7 @@ void handleLocalException(TransportException e) { } } - static final class HandshakeRequest extends TransportRequest { + static final class HandshakeRequest extends AbstractTransportRequest { /** * The {@link TransportVersion#current()} of the requesting node. diff --git a/server/src/main/java/org/elasticsearch/transport/TransportRequest.java b/server/src/main/java/org/elasticsearch/transport/TransportRequest.java index 393d015123473..fda5575e95e7b 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportRequest.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportRequest.java @@ -9,81 +9,14 @@ package org.elasticsearch.transport; -import org.elasticsearch.common.io.stream.StreamInput; -import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.core.Nullable; +import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.core.RefCounted; import org.elasticsearch.tasks.TaskAwareRequest; -import org.elasticsearch.tasks.TaskId; -import java.io.IOException; import java.net.InetSocketAddress; -public abstract class TransportRequest extends TransportMessage implements TaskAwareRequest { +public interface TransportRequest extends Writeable, RefCounted, TaskAwareRequest { + void remoteAddress(InetSocketAddress remoteAddress); - @Nullable // set by the transport service on inbound messages; unset on outbound messages - private InetSocketAddress remoteAddress; - - /** - * Parent of this request. Defaults to {@link TaskId#EMPTY_TASK_ID}, meaning "no parent". - */ - private TaskId parentTaskId = TaskId.EMPTY_TASK_ID; - - /** - * Request ID. Defaults to -1, meaning "no request ID is set". - */ - private volatile long requestId = -1; - - public TransportRequest() {} - - public TransportRequest(StreamInput in) throws IOException { - parentTaskId = TaskId.readFromStream(in); - } - - public void remoteAddress(InetSocketAddress remoteAddress) { - this.remoteAddress = remoteAddress; - } - - @Nullable // set by the transport service on inbound messages; unset on outbound messages - public InetSocketAddress remoteAddress() { - return remoteAddress; - } - - /** - * Set a reference to task that created this request. - */ - @Override - public void setParentTask(TaskId taskId) { - this.parentTaskId = taskId; - } - - /** - * Get a reference to the task that created this request. Defaults to {@link TaskId#EMPTY_TASK_ID}, meaning "there is no parent". - */ - @Override - public TaskId getParentTask() { - return parentTaskId; - } - - /** - * Set the request ID of this request. - */ - @Override - public void setRequestId(long requestId) { - this.requestId = requestId; - } - - @Override - public long getRequestId() { - return requestId; - } - - @Override - public void writeTo(StreamOutput out) throws IOException { - parentTaskId.writeTo(out); - } - - @Override - public String toString() { - return getClass().getName() + "/" + getParentTask(); - } + InetSocketAddress remoteAddress(); } diff --git a/server/src/main/java/org/elasticsearch/transport/TransportService.java b/server/src/main/java/org/elasticsearch/transport/TransportService.java index 01ea7b6a03127..a51184cb08714 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportService.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportService.java @@ -141,7 +141,9 @@ protected boolean removeEldestEntry(Map.Entry eldest) { private final RemoteClusterService remoteClusterService; - /** if set will call requests sent to this id to shortcut and executed locally */ + /** + * if set will call requests sent to this id to shortcut and executed locally + */ volatile DiscoveryNode localNode = null; private final Transport.Connection localNodeConnection = new Transport.Connection() { @Override @@ -220,7 +222,8 @@ public TransportService( * Build the service. * * @param clusterSettings if non null, the {@linkplain TransportService} will register with the {@link ClusterSettings} for settings - * updates for {@link TransportSettings#TRACE_LOG_EXCLUDE_SETTING} and {@link TransportSettings#TRACE_LOG_INCLUDE_SETTING}. + * updates for {@link TransportSettings#TRACE_LOG_EXCLUDE_SETTING} + * and {@link TransportSettings#TRACE_LOG_INCLUDE_SETTING}. */ public TransportService( Settings settings, @@ -424,7 +427,7 @@ protected void doClose() throws IOException { /** * Start accepting incoming requests. - * + *

* The transport service starts before it's ready to accept incoming requests because we need to know the address(es) to which we are * bound, which means we have to actually bind to them and start accepting incoming connections. However until this method is called we * reject any incoming requests, including handshakes, by closing the connection. @@ -484,7 +487,7 @@ public boolean nodeConnected(DiscoveryNode node) { * Connect to the specified node with the given connection profile. * The ActionListener will be called on the calling thread or the generic thread pool. * - * @param node the node to connect to + * @param node the node to connect to * @param listener the action listener to notify */ public void connectToNode(DiscoveryNode node, ActionListener listener) throws ConnectTransportException { @@ -495,9 +498,9 @@ public void connectToNode(DiscoveryNode node, ActionListener listene * Connect to the specified node with the given connection profile. * The ActionListener will be called on the calling thread or the generic thread pool. * - * @param node the node to connect to + * @param node the node to connect to * @param connectionProfile the connection profile to use when connecting to this node - * @param listener the action listener to notify + * @param listener the action listener to notify */ public void connectToNode( final DiscoveryNode node, @@ -540,9 +543,10 @@ public ConnectionManager.ConnectionValidator connectionValidator(DiscoveryNode n * Establishes a new connection to the given node. The connection is NOT maintained by this service, it's the callers * responsibility to close the connection once it goes out of scope. * The ActionListener will be called on the calling thread or the generic thread pool. - * @param node the node to connect to + * + * @param node the node to connect to * @param connectionProfile the connection profile to use - * @param listener the action listener to notify + * @param listener the action listener to notify */ public void openConnection( final DiscoveryNode node, @@ -567,7 +571,7 @@ public void openConnection( * @param handshakeTimeout handshake timeout * @param listener action listener to notify * @throws ConnectTransportException if the connection failed - * @throws IllegalStateException if the handshake failed + * @throws IllegalStateException if the handshake failed */ public void handshake( final Transport.Connection connection, @@ -584,10 +588,10 @@ public void handshake( * name on the target node doesn't match the local cluster name. * The ActionListener will be called on the calling thread or the generic thread pool. * - * @param connection the connection to a specific node - * @param handshakeTimeout handshake timeout + * @param connection the connection to a specific node + * @param handshakeTimeout handshake timeout * @param clusterNamePredicate cluster name validation predicate - * @param listener action listener to notify + * @param listener action listener to notify * @throws IllegalStateException if the handshake failed */ public void handshake( @@ -641,7 +645,7 @@ public RecyclerBytesStreamOutput newNetworkBytesStream() { return transport.newNetworkBytesStream(); } - static class HandshakeRequest extends TransportRequest { + static class HandshakeRequest extends AbstractTransportRequest { public static final HandshakeRequest INSTANCE = new HandshakeRequest(); @@ -884,6 +888,7 @@ private static void handleSendRequestException(TransportResponseHandler handl /** * Returns either a real transport connection or a local node connection if we are using the local node optimization. + * * @throws NodeNotConnectedException if the given node is not connected */ public Transport.Connection getConnection(DiscoveryNode node) { @@ -1189,10 +1194,10 @@ public static boolean isValidActionName(String actionName) { /** * Registers a new request handler * - * @param action The action the request handler is associated with - * @param requestReader a callable to be used construct new instances for streaming - * @param executor The executor the request handling will be executed on - * @param handler The handler itself that implements the request handling + * @param action The action the request handler is associated with + * @param requestReader a callable to be used construct new instances for streaming + * @param executor The executor the request handling will be executed on + * @param handler The handler itself that implements the request handling */ public void registerRequestHandler( String action, @@ -1219,7 +1224,7 @@ public void registerRequestHandler( * Registers a new request handler * * @param action The action the request handler is associated with - * @param requestReader The request class that will be used to construct new instances for streaming + * @param requestReader The request class that will be used to construct new instances for streaming * @param executor The executor the request handling will be executed on * @param forceExecution Force execution on the executor queue and never reject it * @param canTripCircuitBreaker Check the request size and raise an exception in case the limit is breached. @@ -1262,7 +1267,9 @@ public void onRequestReceived(long requestId, String action) { } } - /** called by the {@link Transport} implementation once a request has been sent */ + /** + * called by the {@link Transport} implementation once a request has been sent + */ @Override public void onRequestSent( DiscoveryNode node, @@ -1286,7 +1293,9 @@ public void onResponseReceived(long requestId, Transport.ResponseContext holder) } } - /** called by the {@link Transport} implementation once a response was sent to calling node */ + /** + * called by the {@link Transport} implementation once a response was sent to calling node + */ @Override public void onResponseSent(long requestId, String action) { if (tracerLog.isTraceEnabled() && shouldTraceAction(action)) { @@ -1294,7 +1303,9 @@ public void onResponseSent(long requestId, String action) { } } - /** called by the {@link Transport} implementation after an exception was sent as a response to an incoming request */ + /** + * called by the {@link Transport} implementation after an exception was sent as a response to an incoming request + */ @Override public void onResponseSent(long requestId, String action, Exception e) { if (tracerLog.isTraceEnabled() && shouldTraceAction(action)) { diff --git a/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksTests.java b/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksTests.java index 964dc61688a67..c4fe40628747f 100644 --- a/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksTests.java +++ b/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksTests.java @@ -33,9 +33,9 @@ import org.elasticsearch.tasks.TaskManager; import org.elasticsearch.test.ReachabilityChecker; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.FakeTcpChannel; import org.elasticsearch.transport.TestTransportChannels; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -62,7 +62,7 @@ public class CancellableTasksTests extends TaskManagerTestCase { - public static class CancellableNodeRequest extends TransportRequest { + public static class CancellableNodeRequest extends AbstractTransportRequest { protected String requestName; public CancellableNodeRequest() { diff --git a/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TestTaskPlugin.java b/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TestTaskPlugin.java index 27b48043067d4..8d24bd2ad1bae 100644 --- a/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TestTaskPlugin.java +++ b/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TestTaskPlugin.java @@ -40,6 +40,7 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportInterceptor; @@ -161,7 +162,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { protected final String requestName; protected final boolean shouldBlock; diff --git a/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java b/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java index 3cd0ebb0234b3..1b20d129e4aea 100644 --- a/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java +++ b/server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/TransportTasksActionTests.java @@ -46,7 +46,7 @@ import org.elasticsearch.test.ReachabilityChecker; import org.elasticsearch.test.tasks.MockTaskManager; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xcontent.ToXContent; import org.elasticsearch.xcontent.XContentBuilder; @@ -78,7 +78,7 @@ public class TransportTasksActionTests extends TaskManagerTestCase { - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { protected String requestName; public NodeRequest(StreamInput in) throws IOException { diff --git a/server/src/test/java/org/elasticsearch/action/support/nodes/TransportNodesActionTests.java b/server/src/test/java/org/elasticsearch/action/support/nodes/TransportNodesActionTests.java index 7252e5fe928b1..f954019d5556d 100644 --- a/server/src/test/java/org/elasticsearch/action/support/nodes/TransportNodesActionTests.java +++ b/server/src/test/java/org/elasticsearch/action/support/nodes/TransportNodesActionTests.java @@ -38,8 +38,8 @@ import org.elasticsearch.test.transport.CapturingTransport; import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.LeakTracker; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportService; import org.hamcrest.Matchers; import org.junit.After; @@ -507,7 +507,7 @@ void addCloseListener(ActionListener listener) { } } - private static class TestNodeRequest extends TransportRequest { + private static class TestNodeRequest extends AbstractTransportRequest { private final RefCounted refCounted = AbstractRefCounted.of(() -> {}); TestNodeRequest() {} diff --git a/server/src/test/java/org/elasticsearch/tasks/BanFailureLoggingTests.java b/server/src/test/java/org/elasticsearch/tasks/BanFailureLoggingTests.java index e5cdecd25ef34..9a0cbf6f035e2 100644 --- a/server/src/test/java/org/elasticsearch/tasks/BanFailureLoggingTests.java +++ b/server/src/test/java/org/elasticsearch/tasks/BanFailureLoggingTests.java @@ -25,10 +25,10 @@ import org.elasticsearch.test.transport.StubbableTransport; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.AbstractSimpleTransportTestCase; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.EmptyRequest; import org.elasticsearch.transport.NodeDisconnectedException; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportResponseHandler; @@ -133,7 +133,7 @@ private void runTest( childTransportService.registerRequestHandler( "internal:testAction[c]", threadPool.executor(ThreadPool.Names.MANAGEMENT), // busy-wait for cancellation but not on a transport thread - (StreamInput in) -> new TransportRequest(in) { + (StreamInput in) -> new AbstractTransportRequest(in) { @Override public Task createTask(long id, String type, String action, TaskId parentTaskId, Map headers) { return new CancellableTask(id, type, action, "", parentTaskId, headers); diff --git a/server/src/test/java/org/elasticsearch/tasks/TaskManagerTests.java b/server/src/test/java/org/elasticsearch/tasks/TaskManagerTests.java index a30cb6473f3fa..be7ffdc60d2ea 100644 --- a/server/src/test/java/org/elasticsearch/tasks/TaskManagerTests.java +++ b/server/src/test/java/org/elasticsearch/tasks/TaskManagerTests.java @@ -31,6 +31,7 @@ import org.elasticsearch.test.ESTestCase; import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.FakeTcpChannel; import org.elasticsearch.transport.TcpChannel; import org.elasticsearch.transport.TcpTransportChannel; @@ -42,7 +43,6 @@ import org.elasticsearch.transport.TransportService; import org.junit.After; import org.junit.Before; -import org.mockito.Mockito; import java.util.ArrayList; import java.util.Collections; @@ -282,7 +282,7 @@ public void testTaskAccounting() { * Check that registering a task also causes tracing to be started on that task. */ public void testRegisterTaskStartsTracing() { - final Tracer mockTracer = Mockito.mock(Tracer.class); + final Tracer mockTracer = mock(Tracer.class); final TaskManager taskManager = new TaskManager(Settings.EMPTY, threadPool, Set.of(), mockTracer); final Task task = taskManager.register("testType", "testAction", new TaskAwareRequest() { @@ -306,7 +306,7 @@ public TaskId getParentTask() { * Check that unregistering a task also causes tracing to be stopped on that task. */ public void testUnregisterTaskStopsTracing() { - final Tracer mockTracer = Mockito.mock(Tracer.class); + final Tracer mockTracer = mock(Tracer.class); final TaskManager taskManager = new TaskManager(Settings.EMPTY, threadPool, Set.of(), mockTracer); final Task task = taskManager.register("testType", "testAction", new TaskAwareRequest() { @@ -332,7 +332,7 @@ public TaskId getParentTask() { * Check that registering and executing a task also causes tracing to be started and stopped on that task. */ public void testRegisterAndExecuteStartsAndStopsTracing() { - final Tracer mockTracer = Mockito.mock(Tracer.class); + final Tracer mockTracer = mock(Tracer.class); final TaskManager taskManager = new TaskManager(Settings.EMPTY, threadPool, Set.of(), mockTracer); final Task task = taskManager.registerAndExecute( @@ -370,7 +370,7 @@ public TaskId getParentTask() { } public void testRegisterWithEnabledDisabledTracing() { - final Tracer mockTracer = Mockito.mock(Tracer.class); + final Tracer mockTracer = mock(Tracer.class); final TaskManager taskManager = spy(new TaskManager(Settings.EMPTY, threadPool, Set.of(), mockTracer)); taskManager.register("type", "action", makeTaskRequest(true, 123), false); @@ -390,7 +390,7 @@ public void testRegisterWithEnabledDisabledTracing() { verify(taskManager, times(1)).startTrace(any(), any()); } - static class CancellableRequest extends TransportRequest { + static class CancellableRequest extends AbstractTransportRequest { private final String requestId; CancellableRequest(String requestId) { diff --git a/server/src/test/java/org/elasticsearch/transport/ResultDeduplicatorTests.java b/server/src/test/java/org/elasticsearch/transport/ResultDeduplicatorTests.java index 7cb7f6bc23e86..a75efd56652bd 100644 --- a/server/src/test/java/org/elasticsearch/transport/ResultDeduplicatorTests.java +++ b/server/src/test/java/org/elasticsearch/transport/ResultDeduplicatorTests.java @@ -28,7 +28,7 @@ public void testRequestDeduplication() throws Exception { AtomicInteger successCount = new AtomicInteger(); AtomicInteger failureCount = new AtomicInteger(); Exception failure = randomBoolean() ? new TransportException("simulated") : null; - final TransportRequest request = new TransportRequest() { + final TransportRequest request = new AbstractTransportRequest() { @Override public void setParentTask(final TaskId taskId) {} }; diff --git a/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java b/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java index d49d4882a4965..5bb62f7dd19e6 100644 --- a/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java +++ b/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java @@ -400,7 +400,7 @@ public void handleException(TransportException exp) { latch.await(); } - public static class SimpleTestRequest extends TransportRequest { + public static class SimpleTestRequest extends AbstractTransportRequest { final boolean cancellable; final String sourceNode; diff --git a/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java b/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java index fdc7482a258ab..2fba7b29d0aa3 100644 --- a/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java @@ -246,7 +246,7 @@ private MockTransportService buildService( version, threadPool, clusterSettings, - Collections.emptySet(), + emptySet(), interceptor ); service.start(); @@ -1472,7 +1472,7 @@ public void handleException(TransportException exp) {} } } - public static class StringMessageRequest extends TransportRequest implements RawIndexingDataTransportRequest { + public static class StringMessageRequest extends AbstractTransportRequest implements RawIndexingDataTransportRequest { private String message; private long timeout; @@ -1533,7 +1533,7 @@ public void writeTo(StreamOutput out) throws IOException { } } - public static class Version0Request extends TransportRequest { + public static class Version0Request extends AbstractTransportRequest { int value1; @@ -2053,7 +2053,7 @@ public void handleException(TransportException exp) { } } - public static class TestRequest extends TransportRequest { + public static class TestRequest extends AbstractTransportRequest { String info; int resendCount; @@ -3228,7 +3228,7 @@ public void testChannelToString() { public void testActionStats() throws Exception { final String ACTION = "internal:action"; - class Request extends TransportRequest { + class Request extends AbstractTransportRequest { final int refSize; Request(int refSize) { @@ -3477,7 +3477,7 @@ protected Set getAcceptedChannels(TcpTransport transport) { * Connect to the specified node with the default connection profile * * @param service service to connect from - * @param node the node to connect to + * @param node the node to connect to */ public static void connectToNode(TransportService service, DiscoveryNode node) throws ConnectTransportException { connectToNode(service, node, null); @@ -3486,8 +3486,8 @@ public static void connectToNode(TransportService service, DiscoveryNode node) t /** * Connect to the specified node with the given connection profile * - * @param service service to connect from - * @param node the node to connect to + * @param service service to connect from + * @param node the node to connect to * @param connectionProfile the connection profile to use when connecting to this node */ public static void connectToNode(TransportService service, DiscoveryNode node, ConnectionProfile connectionProfile) { @@ -3512,8 +3512,8 @@ public static ConnectTransportException connectToNodeExpectFailure( /** * Establishes and returns a new connection to the given node from the given {@link TransportService}. * - * @param service service to connect from - * @param node the node to connect to + * @param service service to connect from + * @param node the node to connect to * @param connectionProfile the connection profile to use */ public static Transport.Connection openConnection(TransportService service, DiscoveryNode node, ConnectionProfile connectionProfile) { diff --git a/test/framework/src/main/java/org/elasticsearch/transport/EmptyRequest.java b/test/framework/src/main/java/org/elasticsearch/transport/EmptyRequest.java index cabcca3e8eef8..9da796994aab6 100644 --- a/test/framework/src/main/java/org/elasticsearch/transport/EmptyRequest.java +++ b/test/framework/src/main/java/org/elasticsearch/transport/EmptyRequest.java @@ -17,7 +17,7 @@ * A transport request with an empty payload. Not really entirely empty: all transport requests include the parent task ID, a request ID, * and the remote address (if applicable). */ -public final class EmptyRequest extends TransportRequest { +public final class EmptyRequest extends AbstractTransportRequest { public EmptyRequest() {} public EmptyRequest(StreamInput in) throws IOException { diff --git a/test/framework/src/main/java/org/elasticsearch/transport/TestRequest.java b/test/framework/src/main/java/org/elasticsearch/transport/TestRequest.java index a7807b4ed25c9..cf7867cba6d40 100644 --- a/test/framework/src/main/java/org/elasticsearch/transport/TestRequest.java +++ b/test/framework/src/main/java/org/elasticsearch/transport/TestRequest.java @@ -13,7 +13,7 @@ import java.io.IOException; -public class TestRequest extends TransportRequest { +public class TestRequest extends AbstractTransportRequest { String value; diff --git a/test/framework/src/test/java/org/elasticsearch/transport/DisruptableMockTransportTests.java b/test/framework/src/test/java/org/elasticsearch/transport/DisruptableMockTransportTests.java index 539cb8488bbf6..c6a74fee032d5 100644 --- a/test/framework/src/test/java/org/elasticsearch/transport/DisruptableMockTransportTests.java +++ b/test/framework/src/test/java/org/elasticsearch/transport/DisruptableMockTransportTests.java @@ -620,7 +620,7 @@ public void testBrokenLinkFailsToConnect() { ); } - private class TestRequest extends TransportRequest { + private class TestRequest extends AbstractTransportRequest { private final RefCounted refCounted; TestRequest() { diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/analytics/action/AnalyticsStatsAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/analytics/action/AnalyticsStatsAction.java index 1792ba7aa4637..2034550c5633d 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/analytics/action/AnalyticsStatsAction.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/analytics/action/AnalyticsStatsAction.java @@ -16,7 +16,7 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xcontent.ToXContentObject; import org.elasticsearch.xcontent.XContentBuilder; import org.elasticsearch.xpack.core.common.stats.EnumCounters; @@ -81,7 +81,7 @@ public boolean equals(Object obj) { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { public NodeRequest(StreamInput in) throws IOException { super(in); } @@ -109,9 +109,7 @@ protected void writeNodesTo(StreamOutput out, List nodes) throws I } public EnumCounters getStats() { - List> countersPerNode = getNodes().stream() - .map(AnalyticsStatsAction.NodeResponse::getStats) - .collect(Collectors.toList()); + List> countersPerNode = getNodes().stream().map(NodeResponse::getStats).collect(Collectors.toList()); return EnumCounters.merge(Item.class, countersPerNode); } diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/datatiers/NodesDataTiersUsageTransportAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/datatiers/NodesDataTiersUsageTransportAction.java index 29ceee29127f0..5fe31debec307 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/datatiers/NodesDataTiersUsageTransportAction.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/datatiers/NodesDataTiersUsageTransportAction.java @@ -35,7 +35,7 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -171,7 +171,7 @@ public Task createTask(long id, String type, String action, TaskId parentTaskId, } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { public NodeRequest(StreamInput in) throws IOException { super(in); diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/GetInferenceDiagnosticsAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/GetInferenceDiagnosticsAction.java index db8d1c264ec29..afb59f8d4c843 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/GetInferenceDiagnosticsAction.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/inference/action/GetInferenceDiagnosticsAction.java @@ -18,7 +18,7 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xcontent.ToXContentFragment; import org.elasticsearch.xcontent.ToXContentObject; import org.elasticsearch.xcontent.XContentBuilder; @@ -56,7 +56,7 @@ public int hashCode() { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { public NodeRequest(StreamInput in) throws IOException { super(in); } diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/ClearSecurityCacheRequest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/ClearSecurityCacheRequest.java index 006c2cabe2308..783bb21e68122 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/ClearSecurityCacheRequest.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/ClearSecurityCacheRequest.java @@ -10,7 +10,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -41,7 +41,7 @@ public String[] keys() { return keys; } - public static class Node extends TransportRequest { + public static class Node extends AbstractTransportRequest { private String cacheName; private String[] keys; diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/privilege/ClearPrivilegesCacheRequest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/privilege/ClearPrivilegesCacheRequest.java index 394b5dbab1209..27d2f0d2ff772 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/privilege/ClearPrivilegesCacheRequest.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/privilege/ClearPrivilegesCacheRequest.java @@ -10,7 +10,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -41,7 +41,7 @@ public boolean clearRolesCache() { return clearRolesCache; } - public static class Node extends TransportRequest { + public static class Node extends AbstractTransportRequest { private String[] applicationNames; private boolean clearRolesCache; diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/realm/ClearRealmCacheRequest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/realm/ClearRealmCacheRequest.java index cc35952f494bc..bdad5680bae22 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/realm/ClearRealmCacheRequest.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/realm/ClearRealmCacheRequest.java @@ -9,7 +9,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -23,14 +23,14 @@ public ClearRealmCacheRequest() { } /** - * @return {@code true} if this request targets realms, {@code false} otherwise. + * @return {@code true} if this request targets realms, {@code false} otherwise. */ public boolean allRealms() { return realms == null || realms.length == 0; } /** - * @return The realms that should be evicted. Empty array indicates all realms. + * @return The realms that should be evicted. Empty array indicates all realms. */ public String[] realms() { return realms; @@ -40,7 +40,7 @@ public String[] realms() { * Sets the realms for which caches will be evicted. When not set all the caches of all realms will be * evicted. * - * @param realms The realm names + * @param realms The realm names */ public ClearRealmCacheRequest realms(String... realms) { this.realms = realms; @@ -48,14 +48,14 @@ public ClearRealmCacheRequest realms(String... realms) { } /** - * @return {@code true} if this request targets users, {@code false} otherwise. + * @return {@code true} if this request targets users, {@code false} otherwise. */ public boolean allUsernames() { return usernames == null || usernames.length == 0; } /** - * @return The usernames of the users that should be evicted. Empty array indicates all users. + * @return The usernames of the users that should be evicted. Empty array indicates all users. */ public String[] usernames() { return usernames; @@ -72,7 +72,7 @@ public ClearRealmCacheRequest usernames(String... usernames) { return this; } - public static class Node extends TransportRequest { + public static class Node extends AbstractTransportRequest { private String[] realms; private String[] usernames; diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/role/ClearRolesCacheRequest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/role/ClearRolesCacheRequest.java index 28e75b21a8be8..3b026e149f209 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/role/ClearRolesCacheRequest.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/role/ClearRolesCacheRequest.java @@ -9,7 +9,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -27,7 +27,7 @@ public ClearRolesCacheRequest() { /** * Sets the roles for which caches will be evicted. When not set all the roles will be evicted from the cache. * - * @param names The role names + * @param names The role names */ public ClearRolesCacheRequest names(String... names) { this.names = names; @@ -41,7 +41,7 @@ public String[] names() { return names; } - public static class Node extends TransportRequest { + public static class Node extends AbstractTransportRequest { private String[] names; public Node(StreamInput in) throws IOException { diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/service/GetServiceAccountCredentialsNodesRequest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/service/GetServiceAccountCredentialsNodesRequest.java index c12c830f3712f..8870bae483715 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/service/GetServiceAccountCredentialsNodesRequest.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/service/GetServiceAccountCredentialsNodesRequest.java @@ -10,7 +10,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -29,7 +29,7 @@ public GetServiceAccountCredentialsNodesRequest(String namespace, String service this.serviceName = serviceName; } - public static class Node extends TransportRequest { + public static class Node extends AbstractTransportRequest { private final String namespace; private final String serviceName; diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/spatial/action/SpatialStatsAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/spatial/action/SpatialStatsAction.java index 32dc1550b6aab..6370b7a697e41 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/spatial/action/SpatialStatsAction.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/spatial/action/SpatialStatsAction.java @@ -16,7 +16,7 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xcontent.ToXContentObject; import org.elasticsearch.xcontent.XContentBuilder; import org.elasticsearch.xpack.core.common.stats.EnumCounters; @@ -76,7 +76,7 @@ public boolean equals(Object obj) { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { public NodeRequest(StreamInput in) throws IOException { super(in); } @@ -104,9 +104,7 @@ protected void writeNodesTo(StreamOutput out, List nodes) throws I } public EnumCounters getStats() { - List> countersPerNode = getNodes().stream() - .map(SpatialStatsAction.NodeResponse::getStats) - .collect(Collectors.toList()); + List> countersPerNode = getNodes().stream().map(NodeResponse::getStats).collect(Collectors.toList()); return EnumCounters.merge(Item.class, countersPerNode); } diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/termsenum/action/NodeTermsEnumRequest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/termsenum/action/NodeTermsEnumRequest.java index 55f014db39959..faee6ec12a830 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/termsenum/action/NodeTermsEnumRequest.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/termsenum/action/NodeTermsEnumRequest.java @@ -15,7 +15,7 @@ import org.elasticsearch.core.Nullable; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.index.shard.ShardId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.Collections; @@ -25,7 +25,7 @@ * Internal terms enum request executed directly against a specific node, querying potentially many * shards in one request */ -public class NodeTermsEnumRequest extends TransportRequest implements IndicesRequest { +public class NodeTermsEnumRequest extends AbstractTransportRequest implements IndicesRequest { private final String field; private final String string; diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/GetTransformNodeStatsAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/GetTransformNodeStatsAction.java index 7dd4a7c9d245b..f7d153bf1a1b1 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/GetTransformNodeStatsAction.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/GetTransformNodeStatsAction.java @@ -19,8 +19,7 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.rest.RestStatus; -import org.elasticsearch.transport.TransportRequest; -import org.elasticsearch.xcontent.ToXContent; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xcontent.ToXContentObject; import org.elasticsearch.xcontent.XContentBuilder; import org.elasticsearch.xpack.core.transform.transforms.TransformSchedulerStats; @@ -95,7 +94,7 @@ protected void writeNodesTo(StreamOutput out, List nodes) thr } } - public static class NodeStatsRequest extends TransportRequest { + public static class NodeStatsRequest extends AbstractTransportRequest { public NodeStatsRequest() {} @@ -134,7 +133,7 @@ public void writeTo(StreamOutput out) throws IOException { } @Override - public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException { + public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { builder.startObject(); builder.field(SCHEDULER_STATS_FIELD_NAME, schedulerStats); return builder.endObject(); diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/stats/WatcherStatsRequest.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/stats/WatcherStatsRequest.java index 530b3cde6f90e..061322bb80398 100644 --- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/stats/WatcherStatsRequest.java +++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/transport/actions/stats/WatcherStatsRequest.java @@ -9,7 +9,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -55,7 +55,7 @@ public String toString() { return "watcher_stats"; } - public static class Node extends TransportRequest { + public static class Node extends AbstractTransportRequest { private boolean includeCurrentWatches; private boolean includeQueuedWatches; diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/async/AsyncResultsServiceTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/async/AsyncResultsServiceTests.java index 6849b5b859052..1c69a6a52951a 100644 --- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/async/AsyncResultsServiceTests.java +++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/async/AsyncResultsServiceTests.java @@ -20,7 +20,7 @@ import org.elasticsearch.tasks.TaskId; import org.elasticsearch.tasks.TaskManager; import org.elasticsearch.test.ESSingleNodeTestCase; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.core.async.AsyncSearchIndexServiceTests.TestAsyncResponse; import org.junit.Before; @@ -113,7 +113,7 @@ private synchronized void onFailure(Exception e) { } } - public class TestRequest extends TransportRequest { + public class TestRequest extends AbstractTransportRequest { private final String string; public TestRequest(String string) { diff --git a/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/NodesDeprecationCheckAction.java b/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/NodesDeprecationCheckAction.java index 48009f8105429..e3452d6a94eeb 100644 --- a/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/NodesDeprecationCheckAction.java +++ b/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/NodesDeprecationCheckAction.java @@ -12,7 +12,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xpack.core.deprecation.DeprecationIssue; import java.io.IOException; @@ -31,7 +31,7 @@ private NodesDeprecationCheckAction() { super(NAME); } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { public NodeRequest() {} diff --git a/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/logging/DeprecationCacheResetAction.java b/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/logging/DeprecationCacheResetAction.java index e852ef2f26d77..7405920e5b3cb 100644 --- a/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/logging/DeprecationCacheResetAction.java +++ b/x-pack/plugin/deprecation/src/main/java/org/elasticsearch/xpack/deprecation/logging/DeprecationCacheResetAction.java @@ -18,7 +18,7 @@ import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xcontent.ToXContentObject; import org.elasticsearch.xcontent.XContentBuilder; @@ -27,7 +27,7 @@ import java.util.Objects; /** - Resets deprecation indexing rate limiting cache on each node. + * Resets deprecation indexing rate limiting cache on each node. */ public class DeprecationCacheResetAction extends ActionType { public static final DeprecationCacheResetAction INSTANCE = new DeprecationCacheResetAction(); @@ -96,7 +96,7 @@ public int hashCode() { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { public NodeRequest(StreamInput in) throws IOException { super(in); } diff --git a/x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/action/EnrichCoordinatorStatsAction.java b/x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/action/EnrichCoordinatorStatsAction.java index c38d24392f7e7..d1230fee4d1bc 100644 --- a/x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/action/EnrichCoordinatorStatsAction.java +++ b/x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/action/EnrichCoordinatorStatsAction.java @@ -22,7 +22,7 @@ import org.elasticsearch.injection.guice.Inject; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.core.enrich.action.EnrichStatsAction; import org.elasticsearch.xpack.core.enrich.action.EnrichStatsAction.Response.CoordinatorStats; @@ -51,7 +51,7 @@ public Request() { } } - public static class NodeRequest extends TransportRequest { + public static class NodeRequest extends AbstractTransportRequest { NodeRequest() {} diff --git a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/EqlStatsRequest.java b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/EqlStatsRequest.java index e8753b85d4167..f1506369ba4b2 100644 --- a/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/EqlStatsRequest.java +++ b/x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plugin/EqlStatsRequest.java @@ -10,7 +10,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -38,7 +38,7 @@ public String toString() { return "eql_stats"; } - static class NodeStatsRequest extends TransportRequest { + static class NodeStatsRequest extends AbstractTransportRequest { boolean includeStats; NodeStatsRequest(StreamInput in) throws IOException { diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeRequest.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeRequest.java index 1e8700bcd4030..83012dc653b80 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeRequest.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeRequest.java @@ -12,13 +12,13 @@ import org.elasticsearch.tasks.CancellableTask; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; import java.util.Map; import java.util.Objects; -public final class ExchangeRequest extends TransportRequest { +public final class ExchangeRequest extends AbstractTransportRequest { private final String exchangeId; private final boolean sourcesFinished; diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java index 56536261f834d..b73357524e3f4 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java @@ -33,9 +33,9 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskCancelledException; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportChannel; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportService; @@ -201,7 +201,7 @@ public void finishSessionEarly(String sessionId, ActionListener listener) } } - private static class OpenExchangeRequest extends TransportRequest { + private static class OpenExchangeRequest extends AbstractTransportRequest { private final String sessionId; private final int exchangeBuffer; diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java index 92a27ba9cbefa..34df5c0a6368e 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java @@ -59,6 +59,7 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportChannel; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportRequestOptions; @@ -88,33 +89,33 @@ * page against another index that must have only a single * shard. *

- * This registers a {@link TransportRequestHandler} so we can handle requests - * to join data that isn't local to the node, but it is much faster if the - * data is already local. + * This registers a {@link TransportRequestHandler} so we can handle requests + * to join data that isn't local to the node, but it is much faster if the + * data is already local. *

*

- * The join process spawns a {@link Driver} per incoming page which runs in - * two or three stages: + * The join process spawns a {@link Driver} per incoming page which runs in + * two or three stages: *

*

- * Stage 1: Finding matching document IDs for the input page. This stage is done - * by the {@link EnrichQuerySourceOperator}. The output page of this stage is - * represented as {@code [DocVector, IntBlock: positions of the input terms]}. + * Stage 1: Finding matching document IDs for the input page. This stage is done + * by the {@link EnrichQuerySourceOperator}. The output page of this stage is + * represented as {@code [DocVector, IntBlock: positions of the input terms]}. *

*

- * Stage 2: Extracting field values for the matched document IDs. The output page - * is represented as - * {@code [DocVector, IntBlock: positions, Block: field1, Block: field2,...]}. + * Stage 2: Extracting field values for the matched document IDs. The output page + * is represented as + * {@code [DocVector, IntBlock: positions, Block: field1, Block: field2,...]}. *

*

- * Stage 3: Optionally this combines the extracted values based on positions and filling - * nulls for positions without matches. This is done by {@link MergePositionsOperator}. - * The output page is represented as {@code [Block: field1, Block: field2,...]}. + * Stage 3: Optionally this combines the extracted values based on positions and filling + * nulls for positions without matches. This is done by {@link MergePositionsOperator}. + * The output page is represented as {@code [Block: field1, Block: field2,...]}. *

*

- * The {@link Page#getPositionCount()} of the output {@link Page} is equal to the - * {@link Page#getPositionCount()} of the input page. In other words - it returns - * the same number of rows that it was sent no matter how many documents match. + * The {@link Page#getPositionCount()} of the output {@link Page} is equal to the + * {@link Page#getPositionCount()} of the input page. In other words - it returns + * the same number of rows that it was sent no matter how many documents match. *

*/ public abstract class AbstractLookupService { @@ -478,7 +479,7 @@ abstract static class Request { } } - abstract static class TransportRequest extends org.elasticsearch.transport.TransportRequest implements IndicesRequest { + abstract static class TransportRequest extends AbstractTransportRequest implements IndicesRequest { final String sessionId; final ShardId shardId; /** diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java index 9449e6fe48228..bfb0be4b0212e 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java @@ -24,11 +24,11 @@ import org.elasticsearch.core.Tuple; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.RemoteClusterAware; import org.elasticsearch.transport.RemoteClusterService; import org.elasticsearch.transport.Transport; import org.elasticsearch.transport.TransportChannel; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportResponse; @@ -64,8 +64,8 @@ * Resolves enrich policies across clusters in several steps: * 1. Calculates the policies that need to be resolved for each cluster, see {@link #lookupPolicies}. * 2. Sends out {@link LookupRequest} to each cluster to resolve policies. Internally, a remote cluster handles the lookup in two steps: - * - 2.1 Ensures the caller has permission to access the enrich policies. - * - 2.2 For each found enrich policy, uses {@link IndexResolver} to resolve the mappings of the concrete enrich index. + * - 2.1 Ensures the caller has permission to access the enrich policies. + * - 2.2 For each found enrich policy, uses {@link IndexResolver} to resolve the mappings of the concrete enrich index. * 3. For each unresolved policy, combines the lookup results to compute the actual enrich policy and mappings depending on the enrich mode. * This approach requires at most one cross-cluster call for each cluster. */ @@ -168,6 +168,7 @@ private Collection calculateTargetClusters(Enrich.Mode mode, boolean inc /** * Resolve an enrich policy by merging the lookup responses from the target clusters. + * * @return a resolved enrich policy or an error */ private Tuple mergeLookupResults( @@ -341,7 +342,7 @@ private void failIfSkipUnavailableFalse(Exception e, String cluster, ActionListe } } - private static class LookupRequest extends TransportRequest { + private static class LookupRequest extends AbstractTransportRequest { private final String clusterAlias; private final Collection policyNames; @@ -376,6 +377,7 @@ private static class LookupResponse extends TransportResponse { /** * Use this constructor when the remote cluster is unavailable to indicate inability to do the enrich policy lookup + * * @param connectionError Exception received when trying to connect to a remote cluster */ LookupResponse(Exception connectionError) { diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeRequest.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeRequest.java index 74935e116f064..d83ca4beef52c 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeRequest.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeRequest.java @@ -20,7 +20,7 @@ import org.elasticsearch.tasks.CancellableTask; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput; import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput; import org.elasticsearch.xpack.esql.session.Configuration; @@ -35,7 +35,7 @@ * via {@link ExchangeService#openExchange} before sending this request to the remote cluster. The coordinator on the main cluster * will poll pages from this sink. Internally, this compute will trigger sub-computes on data nodes via {@link DataNodeRequest}. */ -final class ClusterComputeRequest extends TransportRequest implements IndicesRequest.Replaceable { +final class ClusterComputeRequest extends AbstractTransportRequest implements IndicesRequest.Replaceable { private final String clusterAlias; private final String sessionId; private final Configuration configuration; @@ -46,10 +46,10 @@ final class ClusterComputeRequest extends TransportRequest implements IndicesReq /** * A request to start a compute on a remote cluster. * - * @param clusterAlias the cluster alias of this remote cluster - * @param sessionId the sessionId in which the output pages will be placed in the exchange sink specified by this id - * @param configuration the configuration for this compute - * @param plan the physical plan to be executed + * @param clusterAlias the cluster alias of this remote cluster + * @param sessionId the sessionId in which the output pages will be placed in the exchange sink specified by this id + * @param configuration the configuration for this compute + * @param plan the physical plan to be executed */ ClusterComputeRequest(String clusterAlias, String sessionId, Configuration configuration, RemoteClusterPlan plan) { this.clusterAlias = clusterAlias; diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java index 6ebaeca1e56c0..4332e4d6e7054 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java @@ -40,9 +40,9 @@ import org.elasticsearch.tasks.TaskId; import org.elasticsearch.tasks.TaskManager; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.RemoteClusterAware; import org.elasticsearch.transport.TransportException; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.esql.action.EsqlExecutionInfo; import org.elasticsearch.xpack.esql.action.EsqlQueryAction; @@ -75,10 +75,10 @@ /** * Once query is parsed and validated it is scheduled for execution by {@code org.elasticsearch.xpack.esql.plugin.ComputeService#execute} * This method is responsible for splitting physical plan into coordinator and data node plans. - * + *

* Coordinator plan is immediately executed locally (using {@code org.elasticsearch.xpack.esql.plugin.ComputeService#runCompute}) * and is prepared to collect and merge pages from data nodes into the final query result. - * + *

* Data node plan is passed to {@code org.elasticsearch.xpack.esql.plugin.DataNodeComputeHandler#startComputeOnDataNodes} * that is responsible for *

    @@ -510,7 +510,7 @@ CancellableTask createGroupTask(Task parentTask, Supplier description) t } } - private static class ComputeGroupTaskRequest extends TransportRequest { + private static class ComputeGroupTaskRequest extends AbstractTransportRequest { private final Supplier parentDescription; ComputeGroupTaskRequest(TaskId parentTask, Supplier description) { diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequest.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequest.java index 0f113ab78efc0..5b47b6df778dc 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequest.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequest.java @@ -25,8 +25,8 @@ import org.elasticsearch.tasks.CancellableTask; import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.RemoteClusterAware; -import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput; import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput; import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan; @@ -43,7 +43,7 @@ import static org.elasticsearch.xpack.core.security.authz.IndicesAndAliasesResolverField.NO_INDEX_PLACEHOLDER; import static org.elasticsearch.xpack.core.security.authz.IndicesAndAliasesResolverField.NO_INDICES_OR_ALIASES_ARRAY; -final class DataNodeRequest extends TransportRequest implements IndicesRequest.Replaceable { +final class DataNodeRequest extends AbstractTransportRequest implements IndicesRequest.Replaceable { private static final Logger logger = LogManager.getLogger(DataNodeRequest.class); private final String sessionId; diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlStatsRequest.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlStatsRequest.java index 6c1740cac88a5..4b017135e99d9 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlStatsRequest.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlStatsRequest.java @@ -10,7 +10,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -38,7 +38,7 @@ public String toString() { return "esql_stats"; } - static class NodeStatsRequest extends TransportRequest { + static class NodeStatsRequest extends AbstractTransportRequest { boolean includeStats; NodeStatsRequest(StreamInput in) throws IOException { diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportTrainedModelCacheInfoAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportTrainedModelCacheInfoAction.java index c5e4584a80b34..4b0d87720bfdc 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportTrainedModelCacheInfoAction.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportTrainedModelCacheInfoAction.java @@ -18,7 +18,7 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.tasks.TaskId; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.core.ml.action.TrainedModelCacheInfoAction; import org.elasticsearch.xpack.core.ml.action.TrainedModelCacheInfoAction.Response.CacheInfo; @@ -85,7 +85,7 @@ protected CacheInfo nodeOperation(NodeModelCacheInfoRequest nodeModelCacheInfoRe ); } - public static class NodeModelCacheInfoRequest extends TransportRequest { + public static class NodeModelCacheInfoRequest extends AbstractTransportRequest { NodeModelCacheInfoRequest() {} diff --git a/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportClearRepositoriesStatsArchiveAction.java b/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportClearRepositoriesStatsArchiveAction.java index f138449559d25..72eb06eb0a0e4 100644 --- a/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportClearRepositoriesStatsArchiveAction.java +++ b/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportClearRepositoriesStatsArchiveAction.java @@ -19,7 +19,7 @@ import org.elasticsearch.repositories.RepositoryStatsSnapshot; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -78,7 +78,7 @@ protected RepositoriesNodeMeteringResponse nodeOperation(ClearRepositoriesStatsA return new RepositoriesNodeMeteringResponse(clusterService.localNode(), clearedStats); } - static final class ClearRepositoriesStatsArchiveNodeRequest extends TransportRequest { + static final class ClearRepositoriesStatsArchiveNodeRequest extends AbstractTransportRequest { private final long maxVersionToClear; ClearRepositoriesStatsArchiveNodeRequest(long maxVersionToClear) { diff --git a/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportRepositoriesStatsAction.java b/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportRepositoriesStatsAction.java index 76ad89a9dfea9..e9bd9be92a814 100644 --- a/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportRepositoriesStatsAction.java +++ b/x-pack/plugin/repositories-metering-api/src/main/java/org/elasticsearch/xpack/repositories/metering/action/TransportRepositoriesStatsAction.java @@ -17,7 +17,7 @@ import org.elasticsearch.repositories.RepositoriesService; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -75,7 +75,7 @@ protected RepositoriesNodeMeteringResponse nodeOperation(RepositoriesNodeStatsRe return new RepositoriesNodeMeteringResponse(clusterService.localNode(), repositoriesService.repositoriesStats()); } - static final class RepositoriesNodeStatsRequest extends TransportRequest { + static final class RepositoriesNodeStatsRequest extends AbstractTransportRequest { RepositoriesNodeStatsRequest() {} RepositoriesNodeStatsRequest(StreamInput in) throws IOException { diff --git a/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotCacheStoresAction.java b/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotCacheStoresAction.java index 3cbed6874312e..5a685da568ffc 100644 --- a/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotCacheStoresAction.java +++ b/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotCacheStoresAction.java @@ -24,7 +24,7 @@ import org.elasticsearch.snapshots.SnapshotId; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots; import org.elasticsearch.xpack.searchablesnapshots.cache.full.CacheService; @@ -107,7 +107,7 @@ public Request(SnapshotId snapshotId, ShardId shardId, DiscoveryNode[] nodes) { } } - public static final class NodeRequest extends TransportRequest { + public static final class NodeRequest extends AbstractTransportRequest { private final SnapshotId snapshotId; private final ShardId shardId; diff --git a/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotsNodeCachesStatsAction.java b/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotsNodeCachesStatsAction.java index 2ffe523f07e35..cbf4d69015ca5 100644 --- a/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotsNodeCachesStatsAction.java +++ b/x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/cache/TransportSearchableSnapshotsNodeCachesStatsAction.java @@ -26,7 +26,7 @@ import org.elasticsearch.license.XPackLicenseState; import org.elasticsearch.tasks.Task; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportService; import org.elasticsearch.xcontent.ToXContentFragment; import org.elasticsearch.xcontent.ToXContentObject; @@ -135,7 +135,7 @@ protected NodeCachesStatsResponse nodeOperation(NodeRequest request, Task task) ); } - public static final class NodeRequest extends TransportRequest { + public static final class NodeRequest extends AbstractTransportRequest { public NodeRequest() {} diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrailTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrailTests.java index 3be40c280874d..638b1be718040 100644 --- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrailTests.java +++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrailTests.java @@ -39,9 +39,9 @@ import org.elasticsearch.tasks.Task; import org.elasticsearch.telemetry.metric.MeterRegistry; import org.elasticsearch.test.ESTestCase; -import org.elasticsearch.test.rest.FakeRestRequest; import org.elasticsearch.test.rest.FakeRestRequest.Builder; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.xcontent.NamedXContentRegistry; import org.elasticsearch.xcontent.XContentBuilder; @@ -2981,7 +2981,7 @@ protected Tuple prepareRestContent(String uri, InetSocketA private Tuple prepareRestContent(String uri, InetSocketAddress remoteAddress, Map params) { final RestContent content = randomFrom(RestContent.values()); - final FakeRestRequest.Builder builder = new Builder(NamedXContentRegistry.EMPTY); + final Builder builder = new Builder(NamedXContentRegistry.EMPTY); if (content.hasContent()) { builder.withContent(content.content(), XContentType.JSON); } @@ -3005,7 +3005,9 @@ private Tuple prepareRestContent(String uri, InetSocketAdd return new Tuple<>(channel, builder.build()); } - /** creates address without any lookups. hostname can be null, for missing */ + /** + * creates address without any lookups. hostname can be null, for missing + */ protected static InetAddress forge(String hostname, String address) throws IOException { final byte bytes[] = InetAddress.getByName(address).getAddress(); return InetAddress.getByAddress(hostname, bytes); @@ -3054,7 +3056,7 @@ private Authentication createApiKeyAuthenticationAndMaybeWithRunAs(Authenticatio return authentication; } - static class MockRequest extends TransportRequest { + static class MockRequest extends AbstractTransportRequest { MockRequest(ThreadContext threadContext) throws IOException { if (randomBoolean()) { @@ -3265,7 +3267,9 @@ private static void indicesRequest( } } - private record ApiKeyMetadataWithSerialization(Map metadata, String serialization) {}; + private record ApiKeyMetadataWithSerialization(Map metadata, String serialization) {} + + ; private ApiKeyMetadataWithSerialization randomApiKeyMetadataWithSerialization() { final int metadataCase = randomInt(3); @@ -3288,7 +3292,9 @@ private ApiKeyMetadataWithSerialization randomApiKeyMetadataWithSerialization() }; } - private record CrossClusterApiKeyAccessWithSerialization(String access, String serialization) {}; + private record CrossClusterApiKeyAccessWithSerialization(String access, String serialization) {} + + ; private CrossClusterApiKeyAccessWithSerialization randomCrossClusterApiKeyAccessWithSerialization() { return randomFrom( diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/AuthenticationServiceTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/AuthenticationServiceTests.java index 06f6ad2f52f4f..4bf840d281a8c 100644 --- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/AuthenticationServiceTests.java +++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authc/AuthenticationServiceTests.java @@ -43,7 +43,6 @@ import org.elasticsearch.env.Environment; import org.elasticsearch.env.TestEnvironment; import org.elasticsearch.index.get.GetResult; -import org.elasticsearch.index.seqno.SequenceNumbers; import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.license.License; import org.elasticsearch.license.LicensedFeature; @@ -60,6 +59,7 @@ import org.elasticsearch.threadpool.FixedExecutorBuilder; import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.xcontent.NamedXContentRegistry; import org.elasticsearch.xcontent.json.JsonXContent; @@ -557,7 +557,7 @@ public void testAuthenticateSmartRealmOrdering() { // Authenticate against the smart chain. // "SecondRealm" will be at the top of the list and will successfully authc. // "FirstRealm" will not be used - Mockito.reset(operatorPrivilegesService); + reset(operatorPrivilegesService); service.authenticate("_action", transportRequest, true, ActionListener.wrap(result -> { assertThat(expectAuditRequestId(threadContext), is(reqId.get())); assertThat(result, notNullValue()); @@ -699,7 +699,7 @@ public void testAuthenticateSmartRealmOrderingDisabled() { assertTrue(completed.get()); completed.set(false); - Mockito.reset(operatorPrivilegesService); + reset(operatorPrivilegesService); service.authenticate("_action", transportRequest, true, ActionListener.wrap(result -> { assertThat(expectAuditRequestId(threadContext), is(reqId.get())); assertThat(result, notNullValue()); @@ -791,7 +791,7 @@ public void testAuthenticateCached() throws Exception { public void testAuthenticateNonExistentRestRequestUserThrowsAuthenticationException() throws Exception { when(firstRealm.token(threadContext)).thenReturn( - new UsernamePasswordToken("idonotexist", new SecureString("passwd".toCharArray())) + new UsernamePasswordToken("idonotexist", new SecureString("passwd".toCharArray())) ); try { authenticateBlocking(restRequest, null); @@ -1035,7 +1035,7 @@ public void testAuthenticateTransportContextAndHeader() throws Exception { // checking authentication from the context InternalRequest message1 = new InternalRequest(); ThreadPool threadPool1 = new TestThreadPool("testAutheticateTransportContextAndHeader1"); - Mockito.reset(operatorPrivilegesService); + reset(operatorPrivilegesService); try { ThreadContext threadContext1 = threadPool1.getThreadContext(); service = new AuthenticationService( @@ -1079,7 +1079,7 @@ public void testAuthenticateTransportContextAndHeader() throws Exception { // checking authentication from the user header ThreadPool threadPool2 = new TestThreadPool("testAutheticateTransportContextAndHeader2"); - Mockito.reset(operatorPrivilegesService); + reset(operatorPrivilegesService); try { ThreadContext threadContext2 = threadPool2.getThreadContext(); boolean requestIdAlreadyPresent = randomBoolean(); @@ -1243,7 +1243,7 @@ public void testWrongApiKeyDoesNotFallbackToAnonymous() { new GetResult( request.index(), request.id(), - SequenceNumbers.UNASSIGNED_SEQ_NO, + UNASSIGNED_SEQ_NO, UNASSIGNED_PRIMARY_TERM, -1L, false, @@ -2069,23 +2069,23 @@ public void testExpiredToken() throws Exception { when(projectIndex.indexExists()).thenReturn(true); User user = new User("_username", "r1"); final Authentication expected = AuthenticationTestHelper.builder() - .user(user) - .realmRef(new RealmRef("realm", "custom", "node")) - .build(false); + .user(user) + .realmRef(new RealmRef("realm", "custom", "node")) + .build(false); PlainActionFuture tokenFuture = new PlainActionFuture<>(); Tuple newTokenBytes = tokenService.getRandomTokenBytes(randomBoolean()); try (ThreadContext.StoredContext ctx = threadContext.stashContext()) { Authentication originatingAuth = AuthenticationTestHelper.builder() - .user(new User("creator")) - .realmRef(new RealmRef("test", "test", "test")) - .build(false); + .user(new User("creator")) + .realmRef(new RealmRef("test", "test", "test")) + .build(false); tokenService.createOAuth2Tokens( - newTokenBytes.v1(), - newTokenBytes.v2(), - expected, - originatingAuth, - Collections.emptyMap(), - tokenFuture + newTokenBytes.v1(), + newTokenBytes.v2(), + expected, + originatingAuth, + Collections.emptyMap(), + tokenFuture ); } String token = tokenFuture.get().getAccessToken(); @@ -2104,8 +2104,8 @@ public void testExpiredToken() throws Exception { } threadContext.putHeader("Authorization", "Bearer " + token); ElasticsearchSecurityException e = expectThrows( - ElasticsearchSecurityException.class, - () -> authenticateBlocking("_action", transportRequest, null, null) + ElasticsearchSecurityException.class, + () -> authenticateBlocking("_action", transportRequest, null, null) ); if (requestIdAlreadyPresent) { assertThat(expectAuditRequestId(threadContext), is(reqId.get())); @@ -2194,7 +2194,7 @@ public void testApiKeyAuth() { new GetResult( request.index(), request.id(), - SequenceNumbers.UNASSIGNED_SEQ_NO, + UNASSIGNED_SEQ_NO, 1, -1L, false, @@ -2272,7 +2272,7 @@ public void testExpiredApiKey() { new GetResult( request.index(), request.id(), - SequenceNumbers.UNASSIGNED_SEQ_NO, + UNASSIGNED_SEQ_NO, 1, -1L, false, @@ -2306,7 +2306,7 @@ public void testExpiredApiKey() { } public void testCanAuthenticateServiceAccount() { - Mockito.reset(serviceAccountService); + reset(serviceAccountService); final Authentication authentication = AuthenticationTestHelper.builder().serviceAccount().build(); try (ThreadContext.StoredContext ignored = threadContext.newStoredContext()) { boolean requestIdAlreadyPresent = randomBoolean(); @@ -2335,7 +2335,7 @@ public void testCanAuthenticateServiceAccount() { } public void testServiceAccountFailureWillNotFallthrough() throws IOException { - Mockito.reset(serviceAccountService); + reset(serviceAccountService); final ElasticsearchSecurityException bailOut = new ElasticsearchSecurityException("bail out", RestStatus.UNAUTHORIZED); try (ThreadContext.StoredContext ignored = threadContext.newStoredContext()) { boolean requestIdAlreadyPresent = randomBoolean(); @@ -2372,7 +2372,7 @@ public void testServiceAccountFailureWillNotFallthrough() throws IOException { } } - static class InternalRequest extends TransportRequest { + static class InternalRequest extends AbstractTransportRequest { @Override public void writeTo(StreamOutput out) {} } diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizationServiceTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizationServiceTests.java index 904c9af18c7e2..e4bb33c66d983 100644 --- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizationServiceTests.java +++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizationServiceTests.java @@ -132,6 +132,7 @@ import org.elasticsearch.test.ESTestCase; import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.threadpool.ThreadPool.Names; +import org.elasticsearch.transport.AbstractTransportRequest; import org.elasticsearch.transport.EmptyRequest; import org.elasticsearch.transport.TransportActionProxy; import org.elasticsearch.transport.TransportRequest; @@ -154,7 +155,6 @@ import org.elasticsearch.xpack.core.security.authc.Subject; import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine; import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.AuthorizationInfo; -import org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField; import org.elasticsearch.xpack.core.security.authz.IndicesAndAliasesResolverField; import org.elasticsearch.xpack.core.security.authz.ResolvedIndices; import org.elasticsearch.xpack.core.security.authz.RoleDescriptor; @@ -189,7 +189,6 @@ import org.elasticsearch.xpack.sql.action.SqlQueryRequest; import org.junit.Before; import org.mockito.ArgumentMatcher; -import org.mockito.ArgumentMatchers; import org.mockito.Mockito; import java.io.IOException; @@ -1163,9 +1162,7 @@ public void testSearchAgainstEmptyCluster() throws Exception { IndicesOptions.fromOptions(true, true, true, false) ); final ActionListener listener = ActionTestUtils.assertNoFailureListener(ignore -> { - final IndicesAccessControl indicesAccessControl = threadContext.getTransient( - AuthorizationServiceField.INDICES_PERMISSIONS_KEY - ); + final IndicesAccessControl indicesAccessControl = threadContext.getTransient(INDICES_PERMISSIONS_KEY); assertNotNull(indicesAccessControl); final IndicesAccessControl.IndexAccessControl indexAccessControl = indicesAccessControl.getIndexPermissions( IndicesAndAliasesResolverField.NO_INDEX_PLACEHOLDER @@ -1233,16 +1230,16 @@ public void testSearchAgainstIndex() { ); this.setFakeOriginatingAction = false; authorize(authentication, TransportSearchAction.TYPE.name(), searchRequest, true, () -> { - verify(rolesStore).getRoles(Mockito.same(authentication), Mockito.any()); - IndicesAccessControl iac = threadContext.getTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY); + verify(rolesStore).getRoles(Mockito.same(authentication), any()); + IndicesAccessControl iac = threadContext.getTransient(INDICES_PERMISSIONS_KEY); // Successful search action authorization should set a parent authorization header. assertThat(securityContext.getParentAuthorization().action(), equalTo(TransportSearchAction.TYPE.name())); // Within the action handler, execute a child action (the query phase of search) authorize(authentication, SearchTransportService.QUERY_ACTION_NAME, shardRequest, false, () -> { // This child action triggers a second interaction with the role store (which is cached) - verify(rolesStore, times(2)).getRoles(Mockito.same(authentication), Mockito.any()); + verify(rolesStore, times(2)).getRoles(Mockito.same(authentication), any()); // But it does not create a new IndicesAccessControl - assertThat(threadContext.getTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY), sameInstance(iac)); + assertThat(threadContext.getTransient(INDICES_PERMISSIONS_KEY), sameInstance(iac)); // The parent authorization header should only be present for direct child actions // and not be carried over for a child of a child actions. // Meaning, only query phase action should be pre-authorized in this case and potential sub-actions should not. @@ -1298,16 +1295,16 @@ public void testSearchPITAgainstIndex() { ); this.setFakeOriginatingAction = false; authorize(authentication, TransportSearchAction.TYPE.name(), searchRequest, true, () -> { - verify(rolesStore).getRoles(Mockito.same(authentication), Mockito.any()); - IndicesAccessControl iac = threadContext.getTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY); + verify(rolesStore).getRoles(Mockito.same(authentication), any()); + IndicesAccessControl iac = threadContext.getTransient(INDICES_PERMISSIONS_KEY); // Successful search action authorization should set a parent authorization header. assertThat(securityContext.getParentAuthorization().action(), equalTo(TransportSearchAction.TYPE.name())); // Within the action handler, execute a child action (the query phase of search) authorize(authentication, SearchTransportService.QUERY_ACTION_NAME, shardRequest, false, () -> { // This child action triggers a second interaction with the role store (which is cached) - verify(rolesStore, times(2)).getRoles(Mockito.same(authentication), Mockito.any()); + verify(rolesStore, times(2)).getRoles(Mockito.same(authentication), any()); // But it does not create a new IndicesAccessControl - assertThat(threadContext.getTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY), sameInstance(iac)); + assertThat(threadContext.getTransient(INDICES_PERMISSIONS_KEY), sameInstance(iac)); // The parent authorization header should only be present for direct child actions // and not be carried over for a child of a child actions. // Meaning, only query phase action should be pre-authorized in this case and potential sub-actions should not. @@ -2200,7 +2197,7 @@ public void testGrantAllRestrictedUserCannotExecuteOperationAgainstSecurityIndic final SearchRequest searchRequest = new SearchRequest("_all"); authorize(authentication, TransportSearchAction.TYPE.name(), searchRequest); assertEquals(2, searchRequest.indices().length); - assertEquals(IndicesAndAliasesResolverField.NO_INDICES_OR_ALIASES_LIST, Arrays.asList(searchRequest.indices())); + assertEquals(IndicesAndAliasesResolverField.NO_INDICES_OR_ALIASES_LIST, asList(searchRequest.indices())); } public void testMonitoringOperationsAgainstSecurityIndexRequireAllowRestricted() { @@ -2333,7 +2330,7 @@ public void testSuperusersCanExecuteReadOperationAgainstSecurityIndex() { for (final Tuple requestTuple : requests) { final String action = requestTuple.v1(); final TransportRequest request = requestTuple.v2(); - try (ThreadContext.StoredContext ignore = threadContext.newStoredContext()) { + try (StoredContext ignore = threadContext.newStoredContext()) { final Authentication authentication = createAuthentication(superuser); authorize(authentication, action, request); verify(auditTrail).accessGranted( @@ -2411,7 +2408,7 @@ public void testSuperusersCannotExecuteWriteOperationAgainstSecurityIndex() { for (final Tuple requestTuple : requests) { final String action = requestTuple.v1(); final TransportRequest request = requestTuple.v2(); - try (ThreadContext.StoredContext ignore = threadContext.newStoredContext()) { + try (StoredContext ignore = threadContext.newStoredContext()) { final Authentication authentication = createAuthentication(superuser); assertThrowsAuthorizationException( "authentication=[" + authentication + "], action=[" + action + "], request=[" + request + "]", @@ -2577,7 +2574,7 @@ public void testCompositeActionsIndicesAreCheckedAtTheShardLevel() { ); AuditUtil.getOrGenerateRequestId(threadContext); mockEmptyMetadata(); - try (ThreadContext.StoredContext ignore = threadContext.newStoredContext()) { + try (StoredContext ignore = threadContext.newStoredContext()) { authorize(createAuthentication(userAllowed), action, request); } assertThrowsAuthorizationException(() -> authorize(createAuthentication(userDenied), action, request), action, "userDenied"); @@ -2756,7 +2753,7 @@ public void testAuthorizationOfSingleActionMultipleIndicesBulkItems() { mockEmptyMetadata(); final Authentication authentication; final String requestId; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication(new User("user", "good-role")); requestId = AuditUtil.getOrGenerateRequestId(threadContext); authorize(authentication, action, request); @@ -2777,9 +2774,7 @@ public void testAuthorizationOfSingleActionMultipleIndicesBulkItems() { case UPDATE -> TransportUpdateAction.NAME; case DELETE -> TransportDeleteAction.NAME; }), - argThat( - indicesArrays -> indicesArrays.length == allIndexNames.size() && allIndexNames.containsAll(Arrays.asList(indicesArrays)) - ), + argThat(indicesArrays -> indicesArrays.length == allIndexNames.size() && allIndexNames.containsAll(asList(indicesArrays))), eq(BulkItemRequest.class.getSimpleName()), eq(request.remoteAddress()), authzInfoRoles(new String[] { goodRole.getName() }) @@ -2792,7 +2787,7 @@ public void testAuthorizationOfSingleActionMultipleIndicesBulkItems() { final Authentication badAuthentication; final String badRequestId; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { badAuthentication = createAuthentication(new User("bad-user", "bad-role")); badRequestId = AuditUtil.getOrGenerateRequestId(threadContext); // the bulk shard request is authorized, but the bulk items are not @@ -2817,9 +2812,7 @@ public void testAuthorizationOfSingleActionMultipleIndicesBulkItems() { case UPDATE -> TransportUpdateAction.NAME; case DELETE -> TransportDeleteAction.NAME; }), - argThat( - indicesArrays -> indicesArrays.length == allIndexNames.size() && allIndexNames.containsAll(Arrays.asList(indicesArrays)) - ), + argThat(indicesArrays -> indicesArrays.length == allIndexNames.size() && allIndexNames.containsAll(asList(indicesArrays))), eq(BulkItemRequest.class.getSimpleName()), eq(request.remoteAddress()), authzInfoRoles(new String[] { badRole.getName() }) @@ -2886,7 +2879,7 @@ public void testAuthorizationOfMultipleActionsSingleIndexBulkItems() { mockEmptyMetadata(); final Authentication authentication; final String requestId; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication(new User("user", "all-role")); requestId = AuditUtil.getOrGenerateRequestId(threadContext); authorize(authentication, action, request); @@ -2921,7 +2914,7 @@ public void testAuthorizationOfMultipleActionsSingleIndexBulkItems() { // use the "index" role final Authentication indexAuthentication; final String indexRequestId; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { indexAuthentication = createAuthentication(new User("index-user", "index-role")); indexRequestId = AuditUtil.getOrGenerateRequestId(threadContext); authorize(indexAuthentication, action, request); @@ -3142,7 +3135,7 @@ private static Tuple randomCompositeRequest() { }; } - private static class MockCompositeIndicesRequest extends TransportRequest implements CompositeIndicesRequest {} + private static class MockCompositeIndicesRequest extends AbstractTransportRequest implements CompositeIndicesRequest {} private Authentication createAuthentication(User user) { return createAuthentication(user, null); @@ -3471,7 +3464,7 @@ public void loadAuthorizedIndices( RequestInfo requestInfo, AuthorizationInfo authorizationInfo, Map indicesLookup, - ActionListener listener + ActionListener listener ) { throw new UnsupportedOperationException("not implemented"); } @@ -3523,7 +3516,7 @@ public void getUserPrivileges(AuthorizationInfo authorizationInfo, ActionListene projectResolver ); Authentication authentication; - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication(new User("test user", "a_all")); assertEquals(engine, authorizationService.getAuthorizationEngine(authentication)); when(licenseState.isAllowed(Security.AUTHORIZATION_ENGINE_FEATURE)).thenReturn(false); @@ -3531,7 +3524,7 @@ public void getUserPrivileges(AuthorizationInfo authorizationInfo, ActionListene } when(licenseState.isAllowed(Security.AUTHORIZATION_ENGINE_FEATURE)).thenReturn(true); - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication(new User("runas", "runas_role"), new User("runner", "runner_role")); assertEquals(engine, authorizationService.getAuthorizationEngine(authentication)); assertEquals(engine, authorizationService.getRunAsAuthorizationEngine(authentication)); @@ -3541,7 +3534,7 @@ public void getUserPrivileges(AuthorizationInfo authorizationInfo, ActionListene } when(licenseState.isAllowed(Security.AUTHORIZATION_ENGINE_FEATURE)).thenReturn(true); - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication(new User("runas", "runas_role"), new ElasticUser(true)); assertEquals(engine, authorizationService.getAuthorizationEngine(authentication)); assertNotEquals(engine, authorizationService.getRunAsAuthorizationEngine(authentication)); @@ -3552,7 +3545,7 @@ public void getUserPrivileges(AuthorizationInfo authorizationInfo, ActionListene } when(licenseState.isAllowed(Security.AUTHORIZATION_ENGINE_FEATURE)).thenReturn(true); - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication(new User("elastic", "superuser"), new User("runner", "runner_role")); assertNotEquals(engine, authorizationService.getAuthorizationEngine(authentication)); assertThat(authorizationService.getAuthorizationEngine(authentication), instanceOf(RBACEngine.class)); @@ -3563,7 +3556,7 @@ public void getUserPrivileges(AuthorizationInfo authorizationInfo, ActionListene } when(licenseState.isAllowed(Security.AUTHORIZATION_ENGINE_FEATURE)).thenReturn(true); - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication(new User("kibana", "kibana_system"), new ElasticUser(true)); assertNotEquals(engine, authorizationService.getAuthorizationEngine(authentication)); assertThat(authorizationService.getAuthorizationEngine(authentication), instanceOf(RBACEngine.class)); @@ -3575,7 +3568,7 @@ public void getUserPrivileges(AuthorizationInfo authorizationInfo, ActionListene } when(licenseState.isAllowed(Security.AUTHORIZATION_ENGINE_FEATURE)).thenReturn(true); - try (ThreadContext.StoredContext ignore = threadContext.stashContext()) { + try (StoredContext ignore = threadContext.stashContext()) { authentication = createAuthentication( randomFrom(InternalUsers.XPACK_USER, InternalUsers.XPACK_SECURITY_USER, new ElasticUser(true), new KibanaUser(true)) ); @@ -3683,7 +3676,7 @@ public void testRoleRestrictionAccessDenial() { } static AuthorizationInfo authzInfoRoles(String[] expectedRoles) { - return ArgumentMatchers.argThat(new RBACAuthorizationInfoRoleMatcher(expectedRoles)); + return argThat(new RBACAuthorizationInfoRoleMatcher(expectedRoles)); } private static class TestSearchPhaseResult extends SearchPhaseResult { diff --git a/x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlStatsRequest.java b/x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlStatsRequest.java index 3b1b7832f38fd..09b981c6f3b40 100644 --- a/x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlStatsRequest.java +++ b/x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plugin/SqlStatsRequest.java @@ -10,7 +10,7 @@ import org.elasticsearch.action.support.nodes.BaseNodesRequest; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; -import org.elasticsearch.transport.TransportRequest; +import org.elasticsearch.transport.AbstractTransportRequest; import java.io.IOException; @@ -38,7 +38,7 @@ public String toString() { return "sql_stats"; } - static class NodeStatsRequest extends TransportRequest { + static class NodeStatsRequest extends AbstractTransportRequest { boolean includeStats; NodeStatsRequest(StreamInput in) throws IOException {