diff --git a/server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksIT.java b/server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksIT.java index 9c9ebb151d503..65b45580930f1 100644 --- a/server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksIT.java +++ b/server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksIT.java @@ -484,11 +484,6 @@ public void handleException(TransportException exp) { latchedListener.onFailure(exp); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public TestResponse read(StreamInput in) throws IOException { return new TestResponse(in); diff --git a/server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesIndexAction.java b/server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesIndexAction.java index b508e5648bdc7..75a2ed255c4f2 100644 --- a/server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesIndexAction.java +++ b/server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesIndexAction.java @@ -271,11 +271,6 @@ public FieldCapabilitiesIndexResponse read(StreamInput in) throws IOException { return new FieldCapabilitiesIndexResponse(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(final FieldCapabilitiesIndexResponse response) { if (response.canMatch()) { diff --git a/server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java b/server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java index e61ab14092a01..e0e2e8331c607 100644 --- a/server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java +++ b/server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java @@ -54,7 +54,7 @@ public class TransportResyncReplicationAction extends TransportWriteAction implements PrimaryReplicaSyncer.SyncAction { - private static String ACTION_NAME = "internal:index/seq_no/resync"; + private static final String ACTION_NAME = "internal:index/seq_no/resync"; private static final Function EXECUTOR_NAME_FUNCTION = shard -> { if (shard.indexSettings().getIndexMetadata().isSystem()) { return Names.SYSTEM_WRITE; @@ -169,11 +169,6 @@ public ResyncReplicationResponse read(StreamInput in) throws IOException { return newResponseInstance(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(ResyncReplicationResponse response) { final ReplicationResponse.ShardInfo.Failure[] failures = response.getShardInfo().getFailures(); diff --git a/server/src/main/java/org/elasticsearch/action/support/broadcast/TransportBroadcastAction.java b/server/src/main/java/org/elasticsearch/action/support/broadcast/TransportBroadcastAction.java index 6725c92c808c3..87ff49fd718d5 100644 --- a/server/src/main/java/org/elasticsearch/action/support/broadcast/TransportBroadcastAction.java +++ b/server/src/main/java/org/elasticsearch/action/support/broadcast/TransportBroadcastAction.java @@ -183,11 +183,6 @@ public ShardResponse read(StreamInput in) throws IOException { return readShardResponse(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(ShardResponse response) { onOperation(shard, shardIndex, response); 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 d914d6ad3037a..d21489101049d 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 @@ -44,7 +44,6 @@ import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; import org.elasticsearch.tasks.Task; -import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.NodeShouldNotConnectException; import org.elasticsearch.transport.TransportChannel; import org.elasticsearch.transport.TransportException; @@ -336,11 +335,6 @@ public void handleResponse(NodeResponse response) { public void handleException(TransportException exp) { onNodeFailure(node, nodeIndex, exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); } catch (Exception e) { onNodeFailure(node, nodeIndex, e); diff --git a/server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java b/server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java index 1eaa684872b6b..310ae3005539d 100644 --- a/server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java +++ b/server/src/main/java/org/elasticsearch/action/support/nodes/TransportNodesAction.java @@ -226,11 +226,6 @@ public void handleResponse(NodeResponse response) { public void handleException(TransportException exp) { onFailure(idx, node.getId(), exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); } catch (Exception e) { onFailure(idx, nodeId, e); 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 b7f8bc5722710..bfb140af23181 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 @@ -801,11 +801,6 @@ public Response read(StreamInput in) throws IOException { return newResponseInstance(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(Response response) { finishOnSuccess(response); diff --git a/server/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java b/server/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java index 1f2fc5020eb59..9c0ee44d2aa36 100644 --- a/server/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java +++ b/server/src/main/java/org/elasticsearch/action/support/single/instance/TransportInstanceSingleOperationAction.java @@ -198,11 +198,6 @@ public Response read(StreamInput in) throws IOException { return newResponse(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(Response response) { listener.onResponse(response); diff --git a/server/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java b/server/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java index 8acb94ec0b818..e5d8954e57003 100644 --- a/server/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java +++ b/server/src/main/java/org/elasticsearch/action/support/single/shard/TransportSingleShardAction.java @@ -182,11 +182,6 @@ public Response read(StreamInput in) throws IOException { return reader.read(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(final Response response) { listener.onResponse(response); @@ -251,11 +246,6 @@ public Response read(StreamInput in) throws IOException { return reader.read(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(final Response response) { listener.onResponse(response); 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 a20e72d853af5..22485ed873e63 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 @@ -38,7 +38,6 @@ import org.elasticsearch.common.io.stream.Writeable; import org.elasticsearch.common.util.concurrent.AtomicArray; import org.elasticsearch.tasks.Task; -import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.NodeShouldNotConnectException; import org.elasticsearch.transport.TransportChannel; import org.elasticsearch.transport.TransportException; @@ -277,11 +276,6 @@ public void handleResponse(NodeTasksResponse response) { public void handleException(TransportException exp) { onFailure(idx, node.getId(), exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); } } catch (Exception e) { 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 482cff11ea9c5..ad5ca71437104 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java @@ -42,6 +42,7 @@ import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestOptions; import org.elasticsearch.transport.TransportRequestOptions.Type; +import org.elasticsearch.transport.TransportResponse; import org.elasticsearch.transport.TransportResponse.Empty; import org.elasticsearch.transport.TransportResponseHandler; import org.elasticsearch.transport.TransportService; @@ -299,14 +300,10 @@ private void handleWakeUp() { transportService.sendRequest(discoveryNode, FOLLOWER_CHECK_ACTION_NAME, request, TransportRequestOptions.builder().withTimeout(followerCheckTimeout).withType(Type.PING).build(), - new TransportResponseHandler() { - @Override - public Empty read(StreamInput in) { - return Empty.INSTANCE; - } + new TransportResponseHandler.Empty() { @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { if (running() == false) { logger.trace("{} no longer running", FollowerChecker.this); return; @@ -345,12 +342,6 @@ public void handleException(TransportException exp) { failNode(reason); } - - - @Override - public String executor() { - return Names.SAME; - } }); } 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 8983954097849..a8a429a6f9de7 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java @@ -36,7 +36,6 @@ import org.elasticsearch.cluster.service.MasterService; import org.elasticsearch.common.Priority; import org.elasticsearch.common.collect.Tuple; -import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.monitor.NodeHealthService; import org.elasticsearch.monitor.StatusInfo; @@ -240,15 +239,9 @@ public void sendJoinRequest(DiscoveryNode destination, long term, Optional final Tuple dedupKey = Tuple.tuple(destination, joinRequest); if (pendingOutgoingJoins.add(dedupKey)) { logger.debug("attempting to join {} with {}", destination, joinRequest); - transportService.sendRequest(destination, JOIN_ACTION_NAME, joinRequest, - new TransportResponseHandler() { + transportService.sendRequest(destination, JOIN_ACTION_NAME, joinRequest, new TransportResponseHandler.Empty() { @Override - public Empty read(StreamInput in) { - return Empty.INSTANCE; - } - - @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { pendingOutgoingJoins.remove(dedupKey); logger.debug("successfully joined {} with {}", destination, joinRequest); lastFailedJoinAttempt.set(null); @@ -261,11 +254,6 @@ public void handleException(TransportException exp) { attempt.logNow(); lastFailedJoinAttempt.set(attempt); } - - @Override - public String executor() { - return Names.SAME; - } }); } else { logger.debug("already attempting to join {} with request {}, not sending request", destination, joinRequest); @@ -275,15 +263,9 @@ public String executor() { void sendStartJoinRequest(final StartJoinRequest startJoinRequest, final DiscoveryNode destination) { assert startJoinRequest.getSourceNode().isMasterNode() : "sending start-join request for master-ineligible " + startJoinRequest.getSourceNode(); - transportService.sendRequest(destination, START_JOIN_ACTION_NAME, - startJoinRequest, new TransportResponseHandler() { + transportService.sendRequest(destination, START_JOIN_ACTION_NAME, startJoinRequest, new TransportResponseHandler.Empty() { @Override - public Empty read(StreamInput in) { - return Empty.INSTANCE; - } - - @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { logger.debug("successful response to {} from {}", startJoinRequest, destination); } @@ -291,11 +273,6 @@ public void handleResponse(Empty response) { public void handleException(TransportException exp) { logger.debug(new ParameterizedMessage("failure in response to {} from {}", startJoinRequest, destination), exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); } 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 e95acceeda866..2e1789f06acb3 100644 --- a/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java +++ b/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java @@ -218,16 +218,10 @@ void handleWakeUp() { transportService.sendRequest(leader, LEADER_CHECK_ACTION_NAME, new LeaderCheckRequest(transportService.getLocalNode()), TransportRequestOptions.builder().withTimeout(leaderCheckTimeout).withType(Type.PING).build(), - - new TransportResponseHandler() { - - @Override - public Empty read(StreamInput in) { - return Empty.INSTANCE; - } + new TransportResponseHandler.Empty() { @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { if (isClosed.get()) { logger.debug("closed check scheduler received a response, doing nothing"); return; @@ -269,11 +263,6 @@ public void handleException(TransportException exp) { failureCount, LEADER_CHECK_RETRY_COUNT_SETTING.getKey(), leaderCheckRetryCount, leader), exp); scheduleNextWakeUp(); } - - @Override - public String executor() { - return Names.SAME; - } }); } diff --git a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java index 3e003fc2898f9..41b821f077246 100644 --- a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java +++ b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java @@ -118,11 +118,6 @@ public ReplicationResponse read(StreamInput in) throws IOException { return newResponseInstance(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(ReplicationResponse response) { task.setPhase("finished"); diff --git a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseSyncAction.java b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseSyncAction.java index 67da626db6e2f..d4f7cb0ea05e0 100644 --- a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseSyncAction.java +++ b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseSyncAction.java @@ -120,11 +120,6 @@ public ReplicationResponse read(StreamInput in) throws IOException { return newResponseInstance(in); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public void handleResponse(ReplicationResponse response) { task.setPhase("finished"); 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 1ee736dce7684..e601698236fd0 100644 --- a/server/src/main/java/org/elasticsearch/indices/store/IndicesStore.java +++ b/server/src/main/java/org/elasticsearch/indices/store/IndicesStore.java @@ -260,11 +260,6 @@ public void handleException(TransportException exp) { } } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - private void allNodesResponded() { if (activeCopies.get() != expectedActiveCopies) { logger.trace("not deleting shard {}, expected {} active copies, but only {} found active copies", diff --git a/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java b/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java index 4a958ecda070c..d972ad7d5838b 100644 --- a/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java +++ b/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java @@ -467,11 +467,6 @@ public void handleResponse(UpdateIndexShardSnapshotStatusResponse response) { public void handleException(TransportException exp) { reqListener.onFailure(exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }) ); } diff --git a/server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java b/server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java index be663117154fb..da769e1681bd8 100644 --- a/server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java +++ b/server/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java @@ -146,11 +146,6 @@ public void handleResponse(ClusterStateResponse response) { public void handleException(TransportException exp) { contextPreservingActionListener.onFailure(exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); } }; diff --git a/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java b/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java index ff351743b5999..beb5cfb8084d1 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java @@ -92,11 +92,6 @@ public void handleException(TransportException exp) { throw new UncheckedIOException(e); } } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } } static class ProxyRequest extends TransportRequest { diff --git a/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java b/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java index 20e6bf4aa79a3..ee23957cc0cac 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java @@ -151,11 +151,6 @@ void handleLocalException(TransportException e) { listener.onFailure(e); } } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } } static final class HandshakeRequest extends TransportRequest { diff --git a/server/src/main/java/org/elasticsearch/transport/TransportResponseHandler.java b/server/src/main/java/org/elasticsearch/transport/TransportResponseHandler.java index 29720216cf400..7eb325e53bfad 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportResponseHandler.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportResponseHandler.java @@ -19,7 +19,9 @@ package org.elasticsearch.transport; +import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.threadpool.ThreadPool; public interface TransportResponseHandler extends Writeable.Reader { @@ -27,5 +29,17 @@ public interface TransportResponseHandler extends W void handleException(TransportException exp); - String executor(); + default String executor() { + return ThreadPool.Names.SAME; + } + + /** + * Implementations of {@link TransportResponseHandler} that handles the empty response {@link TransportResponse.Empty}. + */ + abstract class Empty implements TransportResponseHandler { + @Override + public final TransportResponse.Empty read(StreamInput in) { + return TransportResponse.Empty.INSTANCE; + } + } } diff --git a/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportAddVotingConfigExclusionsActionTests.java b/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportAddVotingConfigExclusionsActionTests.java index 1343c30749b20..39581128d0900 100644 --- a/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportAddVotingConfigExclusionsActionTests.java +++ b/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportAddVotingConfigExclusionsActionTests.java @@ -46,7 +46,6 @@ import org.elasticsearch.test.transport.MockTransport; import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.threadpool.ThreadPool.Names; import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportResponseHandler; import org.elasticsearch.transport.TransportService; @@ -440,11 +439,6 @@ public void handleException(TransportException exp) { onException.accept(exp); } - @Override - public String executor() { - return Names.SAME; - } - @Override public AddVotingConfigExclusionsResponse read(StreamInput in) throws IOException { return new AddVotingConfigExclusionsResponse(in); diff --git a/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportClearVotingConfigExclusionsActionTests.java b/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportClearVotingConfigExclusionsActionTests.java index e7fd68b6c0d0a..b7fd80018eef9 100644 --- a/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportClearVotingConfigExclusionsActionTests.java +++ b/server/src/test/java/org/elasticsearch/action/admin/cluster/configuration/TransportClearVotingConfigExclusionsActionTests.java @@ -40,7 +40,6 @@ import org.elasticsearch.test.transport.MockTransport; import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.threadpool.ThreadPool.Names; import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportResponseHandler; import org.elasticsearch.transport.TransportService; @@ -200,11 +199,6 @@ public void handleException(TransportException exp) { onException.accept(exp); } - @Override - public String executor() { - return Names.SAME; - } - @Override public ClearVotingConfigExclusionsResponse read(StreamInput in) throws IOException { return new ClearVotingConfigExclusionsResponse(in); diff --git a/server/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationAllPermitsAcquisitionTests.java b/server/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationAllPermitsAcquisitionTests.java index 901bd28cb7591..2b3a423a6ccf2 100644 --- a/server/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationAllPermitsAcquisitionTests.java +++ b/server/src/test/java/org/elasticsearch/action/support/replication/TransportReplicationAllPermitsAcquisitionTests.java @@ -203,11 +203,6 @@ public void handleResponse(TransportReplicationAction.ReplicaResponse response) public void handleException(TransportException exp) { getResponseHandler().handleException(exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); } }; diff --git a/server/src/test/java/org/elasticsearch/cluster/coordination/FollowersCheckerTests.java b/server/src/test/java/org/elasticsearch/cluster/coordination/FollowersCheckerTests.java index f2400e1fc2f33..329332b6afd42 100644 --- a/server/src/test/java/org/elasticsearch/cluster/coordination/FollowersCheckerTests.java +++ b/server/src/test/java/org/elasticsearch/cluster/coordination/FollowersCheckerTests.java @@ -26,7 +26,6 @@ import org.elasticsearch.cluster.node.DiscoveryNode; import org.elasticsearch.cluster.node.DiscoveryNodeRole; import org.elasticsearch.cluster.node.DiscoveryNodes; -import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings.Builder; import org.elasticsearch.monitor.NodeHealthService; @@ -36,7 +35,6 @@ import org.elasticsearch.test.EqualsHashCodeTestUtils.CopyFunction; import org.elasticsearch.test.transport.CapturingTransport; import org.elasticsearch.test.transport.MockTransport; -import org.elasticsearch.threadpool.ThreadPool.Names; import org.elasticsearch.transport.AbstractSimpleTransportTestCase; import org.elasticsearch.transport.ConnectTransportException; import org.elasticsearch.transport.TransportException; @@ -425,12 +423,7 @@ protected void onSendRequest(long requestId, String action, TransportRequest req followersChecker.updateFastResponseState(followerTerm, Mode.FOLLOWER); final AtomicReference receivedException = new AtomicReference<>(); transportService.sendRequest(follower, FOLLOWER_CHECK_ACTION_NAME, new FollowerCheckRequest(leaderTerm, leader), - new TransportResponseHandler() { - @Override - public TransportResponse.Empty read(StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } - + new TransportResponseHandler.Empty() { @Override public void handleResponse(TransportResponse.Empty response) { fail("unexpected success"); @@ -441,11 +434,6 @@ public void handleException(TransportException exp) { assertThat(exp, not(nullValue())); assertTrue(receivedException.compareAndSet(null, exp)); } - - @Override - public String executor() { - return Names.SAME; - } }); deterministicTaskQueue.runAllTasks(); assertFalse(calledCoordinator.get()); @@ -504,12 +492,7 @@ protected void onSendRequest(long requestId, String action, TransportRequest req final AtomicReference receivedException = new AtomicReference<>(); transportService.sendRequest(follower, FOLLOWER_CHECK_ACTION_NAME, new FollowerCheckRequest(leaderTerm, leader), - new TransportResponseHandler() { - @Override - public TransportResponse.Empty read(StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } - + new TransportResponseHandler.Empty() { @Override public void handleResponse(TransportResponse.Empty response) { fail("unexpected success"); @@ -520,11 +503,6 @@ public void handleException(TransportException exp) { assertThat(exp, not(nullValue())); assertTrue(receivedException.compareAndSet(null, exp)); } - - @Override - public String executor() { - return Names.SAME; - } }); deterministicTaskQueue.runAllTasks(); assertFalse(calledCoordinator.get()); @@ -568,12 +546,7 @@ public String executor() { final AtomicReference receivedException = new AtomicReference<>(); transportService.sendRequest(follower, FOLLOWER_CHECK_ACTION_NAME, new FollowerCheckRequest(term, leader), - new TransportResponseHandler() { - @Override - public TransportResponse.Empty read(StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } - + new TransportResponseHandler.Empty() { @Override public void handleResponse(TransportResponse.Empty response) { fail("unexpected success"); @@ -584,11 +557,6 @@ public void handleException(TransportException exp) { assertThat(exp, not(nullValue())); assertTrue(receivedException.compareAndSet(null, exp)); } - - @Override - public String executor() { - return Names.SAME; - } }); deterministicTaskQueue.runAllTasks(); assertTrue(calledCoordinator.get()); @@ -653,11 +621,11 @@ private static Settings randomSettings() { return settingsBuilder.build(); } - private static class ExpectsSuccess implements TransportResponseHandler { + private static class ExpectsSuccess extends TransportResponseHandler.Empty { private final AtomicBoolean responseReceived = new AtomicBoolean(); @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { assertTrue(responseReceived.compareAndSet(false, true)); } @@ -666,19 +634,9 @@ public void handleException(TransportException exp) { throw new AssertionError("unexpected", exp); } - @Override - public String executor() { - return Names.SAME; - } - public boolean succeeded() { return responseReceived.get(); } - @Override - public TransportResponse.Empty read(StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } - } } diff --git a/server/src/test/java/org/elasticsearch/cluster/coordination/LeaderCheckerTests.java b/server/src/test/java/org/elasticsearch/cluster/coordination/LeaderCheckerTests.java index 33791c822c6ab..b4911fc62fc0a 100644 --- a/server/src/test/java/org/elasticsearch/cluster/coordination/LeaderCheckerTests.java +++ b/server/src/test/java/org/elasticsearch/cluster/coordination/LeaderCheckerTests.java @@ -465,13 +465,13 @@ public void testLeaderBehaviour() { } } - private class CapturingTransportResponseHandler implements TransportResponseHandler { + private class CapturingTransportResponseHandler implements TransportResponseHandler { TransportException transportException; boolean successfulResponseReceived; @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { successfulResponseReceived = true; } diff --git a/server/src/test/java/org/elasticsearch/cluster/coordination/PreVoteCollectorTests.java b/server/src/test/java/org/elasticsearch/cluster/coordination/PreVoteCollectorTests.java index 8026dcc14779f..420bb2fd0c27d 100644 --- a/server/src/test/java/org/elasticsearch/cluster/coordination/PreVoteCollectorTests.java +++ b/server/src/test/java/org/elasticsearch/cluster/coordination/PreVoteCollectorTests.java @@ -50,7 +50,6 @@ import static org.elasticsearch.monitor.StatusInfo.Status.HEALTHY; import static org.elasticsearch.monitor.StatusInfo.Status.UNHEALTHY; import static org.elasticsearch.node.Node.NODE_NAME_SETTING; -import static org.elasticsearch.threadpool.ThreadPool.Names.SAME; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; @@ -297,11 +296,6 @@ public void handleResponse(PreVoteResponse response) { public void handleException(TransportException exp) { exceptionRef.set(exp); } - - @Override - public String executor() { - return SAME; - } }); deterministicTaskQueue.runAllRunnableTasks(); diff --git a/server/src/test/java/org/elasticsearch/discovery/PeerFinderTests.java b/server/src/test/java/org/elasticsearch/discovery/PeerFinderTests.java index 1c268233d6cc3..e17f947fe9a17 100644 --- a/server/src/test/java/org/elasticsearch/discovery/PeerFinderTests.java +++ b/server/src/test/java/org/elasticsearch/discovery/PeerFinderTests.java @@ -35,7 +35,6 @@ import org.elasticsearch.test.transport.CapturingTransport; import org.elasticsearch.test.transport.CapturingTransport.CapturedRequest; import org.elasticsearch.test.transport.StubbableConnectionManager; -import org.elasticsearch.threadpool.ThreadPool.Names; import org.elasticsearch.transport.ClusterConnectionManager; import org.elasticsearch.transport.ConnectionManager; import org.elasticsearch.transport.TransportException; @@ -507,11 +506,6 @@ public void handleResponse(PeersResponse response) { public void handleException(TransportException exp) { throw new AssertionError("unexpected", exp); } - - @Override - public String executor() { - return Names.SAME; - } }); runAllRunnableTasks(); diff --git a/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java b/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java index 1a8c29ee173af..a64bc9aa5df76 100644 --- a/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java +++ b/server/src/test/java/org/elasticsearch/transport/InboundHandlerTests.java @@ -123,11 +123,6 @@ public void handleException(TransportException exp) { exceptionCaptor.set(exp); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public TestResponse read(StreamInput in) throws IOException { return new TestResponse(in); diff --git a/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java b/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java index 24e6a2579dca3..0e71b8427b712 100644 --- a/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java +++ b/server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java @@ -133,11 +133,6 @@ public void handleException(TransportException exp) { latch.countDown(); } } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); latch.await(); } @@ -192,11 +187,6 @@ public void handleException(TransportException exp) { latch.countDown(); } } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); latch.await(); } diff --git a/server/src/test/java/org/elasticsearch/transport/TransportServiceDeserializationFailureTests.java b/server/src/test/java/org/elasticsearch/transport/TransportServiceDeserializationFailureTests.java index be862ef45f592..d7e18d46cb150 100644 --- a/server/src/test/java/org/elasticsearch/transport/TransportServiceDeserializationFailureTests.java +++ b/server/src/test/java/org/elasticsearch/transport/TransportServiceDeserializationFailureTests.java @@ -91,11 +91,6 @@ public void handleException(TransportException exp) { fail("should not be called"); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public TransportResponse.Empty read(StreamInput in) { throw new AssertionError("should not be called"); @@ -141,11 +136,6 @@ public void handleException(TransportException exp) { fail("should not be called"); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public TransportResponse.Empty read(StreamInput in) { throw new AssertionError("should not be called"); 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 417e1667ddb71..77f33f330a2a3 100644 --- a/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java @@ -1042,11 +1042,6 @@ public void handleResponse(StringMessageResponse response) { @Override public void handleException(TransportException exp) { } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }; serviceA.registerRequestHandler("internal:test", ThreadPool.Names.SAME, StringMessageRequest::new, handler); @@ -1314,11 +1309,6 @@ public void handleException(TransportException exp) { logger.error("Unexpected failure", exp); fail("got exception instead of a response: " + exp.getMessage()); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }).txGet(); assertThat(version0Response.value1, equalTo(1)); @@ -1357,11 +1347,6 @@ public void handleException(TransportException exp) { logger.error("Unexpected failure", exp); fail("got exception instead of a response: " + exp.getMessage()); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }).txGet(); assertThat(version1Response.value1, equalTo(1)); @@ -1399,11 +1384,6 @@ public void handleException(TransportException exp) { logger.error("Unexpected failure", exp); fail("got exception instead of a response: " + exp.getMessage()); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }).txGet(); assertThat(version1Response.value1, equalTo(1)); @@ -1438,11 +1418,6 @@ public void handleException(TransportException exp) { logger.error("Unexpected failure", exp); fail("got exception instead of a response: " + exp.getMessage()); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }).txGet(); assertThat(version0Response.value1, equalTo(1)); @@ -1582,11 +1557,6 @@ public void handleResponse(TestResponse response) { public void handleException(TransportException exp) { latch.countDown(); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); if (!latch.await(10, TimeUnit.SECONDS)) { @@ -1629,11 +1599,6 @@ public void handleResponse(TestResponse response) { public void handleException(TransportException exp) { latch.countDown(); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); latch.await(); @@ -1659,11 +1624,6 @@ public void handleResponse(TestResponse response) { public void handleException(TransportException exp) { latch2.countDown(); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }); latch2.await(); @@ -2241,14 +2201,9 @@ protected void doRun() throws Exception { serviceC.start(); serviceC.acceptIncomingRequests(); CountDownLatch responseLatch = new CountDownLatch(1); - TransportResponseHandler transportResponseHandler = new TransportResponseHandler() { - @Override - public TransportResponse read(StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } - + TransportResponseHandler transportResponseHandler = new TransportResponseHandler.Empty() { @Override - public void handleResponse(TransportResponse response) { + public void handleResponse(TransportResponse.Empty response) { responseLatch.countDown(); } @@ -2256,11 +2211,6 @@ public void handleResponse(TransportResponse response) { public void handleException(TransportException exp) { responseLatch.countDown(); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }; ConnectionProfile.Builder builder = new ConnectionProfile.Builder(); @@ -2309,14 +2259,9 @@ protected void doRun() throws Exception { serviceC.start(); serviceC.acceptIncomingRequests(); CountDownLatch responseLatch = new CountDownLatch(1); - TransportResponseHandler transportResponseHandler = new TransportResponseHandler() { - @Override - public TransportResponse read(StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } - + TransportResponseHandler transportResponseHandler = new TransportResponseHandler.Empty() { @Override - public void handleResponse(TransportResponse response) { + public void handleResponse(TransportResponse.Empty response) { responseLatch.countDown(); } @@ -2324,11 +2269,6 @@ public void handleResponse(TransportResponse response) { public void handleException(TransportException exp) { responseLatch.countDown(); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }; TransportStats stats = serviceC.transport.getStats(); // nothing transmitted / read yet @@ -2423,14 +2363,9 @@ protected void doRun() throws Exception { serviceC.acceptIncomingRequests(); CountDownLatch responseLatch = new CountDownLatch(1); AtomicReference receivedException = new AtomicReference<>(null); - TransportResponseHandler transportResponseHandler = new TransportResponseHandler() { + TransportResponseHandler transportResponseHandler = new TransportResponseHandler.Empty() { @Override - public TransportResponse read(StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } - - @Override - public void handleResponse(TransportResponse response) { + public void handleResponse(TransportResponse.Empty response) { responseLatch.countDown(); } @@ -2439,11 +2374,6 @@ public void handleException(TransportException exp) { receivedException.set(exp); responseLatch.countDown(); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }; TransportStats stats = serviceC.transport.getStats(); // nothing transmitted / read yet @@ -2789,9 +2719,9 @@ public void onFailure(final Exception e) { "fail-to-send-action", TransportRequest.Empty.INSTANCE, TransportRequestOptions.EMPTY, - new TransportResponseHandler() { + new TransportResponseHandler.Empty() { @Override - public void handleResponse(final TransportResponse response) { + public void handleResponse(final TransportResponse.Empty response) { fail("handle response should not be invoked"); } @@ -2799,16 +2729,6 @@ public void handleResponse(final TransportResponse response) { public void handleException(final TransportException exp) { te.set(exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - - @Override - public TransportResponse read(final StreamInput in) { - return TransportResponse.Empty.INSTANCE; - } }); assertThat(te.get(), not(nullValue())); diff --git a/test/framework/src/test/java/org/elasticsearch/test/disruption/DisruptableMockTransportTests.java b/test/framework/src/test/java/org/elasticsearch/test/disruption/DisruptableMockTransportTests.java index 5f7f2085a2c40..2d11ef0a40609 100644 --- a/test/framework/src/test/java/org/elasticsearch/test/disruption/DisruptableMockTransportTests.java +++ b/test/framework/src/test/java/org/elasticsearch/test/disruption/DisruptableMockTransportTests.java @@ -38,7 +38,6 @@ import org.elasticsearch.transport.TransportRequest; import org.elasticsearch.transport.TransportRequestHandler; import org.elasticsearch.transport.TransportResponse; -import org.elasticsearch.transport.TransportResponse.Empty; import org.elasticsearch.transport.TransportResponseHandler; import org.elasticsearch.transport.TransportService; import org.junit.Before; @@ -186,7 +185,7 @@ private TransportRequestHandler requestHandlerCaptures(C } private TransportResponseHandler responseHandlerShouldNotBeCalled() { - return new TransportResponseHandler() { + return new TransportResponseHandler<>() { @Override public TransportResponse read(StreamInput in) { throw new AssertionError("should not be called"); @@ -201,23 +200,13 @@ public void handleResponse(TransportResponse response) { public void handleException(TransportException exp) { throw new AssertionError("should not be called"); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }; } - private TransportResponseHandler responseHandlerShouldBeCalledNormally(Runnable onCalled) { - return new TransportResponseHandler() { - @Override - public TransportResponse read(StreamInput in) { - return Empty.INSTANCE; - } - + private TransportResponseHandler responseHandlerShouldBeCalledNormally(Runnable onCalled) { + return new TransportResponseHandler.Empty() { @Override - public void handleResponse(TransportResponse response) { + public void handleResponse(TransportResponse.Empty response) { onCalled.run(); } @@ -225,16 +214,11 @@ public void handleResponse(TransportResponse response) { public void handleException(TransportException exp) { throw new AssertionError("should not be called"); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }; } private TransportResponseHandler responseHandlerShouldBeCalledExceptionally(Consumer onCalled) { - return new TransportResponseHandler() { + return new TransportResponseHandler<>() { @Override public TransportResponse read(StreamInput in) { throw new AssertionError("should not be called"); @@ -249,11 +233,6 @@ public void handleResponse(TransportResponse response) { public void handleException(TransportException exp) { onCalled.accept(exp); } - - @Override - public String executor() { - return ThreadPool.Names.SAME; - } }; } @@ -262,7 +241,7 @@ private void registerRequestHandler(TransportService transportService, Transport } private void send(TransportService transportService, DiscoveryNode destinationNode, - TransportResponseHandler responseHandler) { + TransportResponseHandler responseHandler) { transportService.sendRequest(destinationNode, "internal:dummy", TransportRequest.Empty.INSTANCE, responseHandler); } diff --git a/test/framework/src/test/java/org/elasticsearch/test/disruption/NetworkDisruptionIT.java b/test/framework/src/test/java/org/elasticsearch/test/disruption/NetworkDisruptionIT.java index c335572acb567..b0abdbddfe38a 100644 --- a/test/framework/src/test/java/org/elasticsearch/test/disruption/NetworkDisruptionIT.java +++ b/test/framework/src/test/java/org/elasticsearch/test/disruption/NetworkDisruptionIT.java @@ -31,7 +31,6 @@ import org.elasticsearch.test.InternalTestCluster; import org.elasticsearch.test.disruption.NetworkDisruption.TwoPartitions; import org.elasticsearch.test.transport.MockTransportService; -import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportResponse; import org.elasticsearch.transport.TransportResponseHandler; @@ -178,11 +177,6 @@ public void handleException(TransportException exp) { latch.countDown(); } - @Override - public String executor() { - return ThreadPool.Names.SAME; - } - @Override public TransportResponse read(StreamInput in) throws IOException { return ClusterHealthResponse.readResponseFrom(in); diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java index 9c832ab59bb5c..d2019af7e03e6 100644 --- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java +++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java @@ -11,7 +11,6 @@ import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.cluster.block.ClusterBlocks; import org.elasticsearch.cluster.service.ClusterService; -import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.settings.ClusterSettings; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.util.concurrent.ThreadContext; @@ -369,15 +368,9 @@ public void testContextRestoreResponseHandler() throws Exception { threadContext.putTransient("foo", "different_bar"); threadContext.putHeader("key", "value2"); TransportResponseHandler handler = new TransportService.ContextRestoreResponseHandler<>( - threadContext.wrapRestorable(storedContext), new TransportResponseHandler() { - - @Override - public Empty read(StreamInput in) { - return Empty.INSTANCE; - } - + threadContext.wrapRestorable(storedContext), new TransportResponseHandler.Empty() { @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { assertEquals("bar", threadContext.getTransient("foo")); assertEquals("value", threadContext.getHeader("key")); } @@ -387,11 +380,6 @@ public void handleException(TransportException exp) { assertEquals("bar", threadContext.getTransient("foo")); assertEquals("value", threadContext.getHeader("key")); } - - @Override - public String executor() { - return null; - } }); handler.handleResponse(null); @@ -408,15 +396,9 @@ public void testContextRestoreResponseHandlerRestoreOriginalContext() throws Exc threadContext.putTransient("foo", "different_bar"); threadContext.putHeader("key", "value2"); handler = new TransportService.ContextRestoreResponseHandler<>(threadContext.newRestorableContext(true), - new TransportResponseHandler() { - - @Override - public Empty read(StreamInput in) { - return Empty.INSTANCE; - } - + new TransportResponseHandler.Empty() { @Override - public void handleResponse(Empty response) { + public void handleResponse(TransportResponse.Empty response) { assertEquals("different_bar", threadContext.getTransient("foo")); assertEquals("value2", threadContext.getHeader("key")); } @@ -426,11 +408,6 @@ public void handleException(TransportException exp) { assertEquals("different_bar", threadContext.getTransient("foo")); assertEquals("value2", threadContext.getHeader("key")); } - - @Override - public String executor() { - return null; - } }); }