Move more work that blocks on PeerFinder mutex to coordinating thread (#95368)

follow up to #95341 

Moving a couple of things that eventually use the peer finder and
involve cluster coordination work to the coordinating thread. This
avoids blocking on its mutex on the scheduler and transport threads and
causing needless contention by running work on the generic pool that can
very well wait for its turn on the cluster coordinating pool. Also this
takes some more load off of the transport threads by moving response
serializations to the coordinating pool.

Author: original-brownbear

server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationDiagnosticsService.java

@@ -1111,7 +1111,8 @@ private <R extends ActionResponse, T> Scheduler.Cancellable sendTransportRequest
                     TransportRequestOptions.timeout(transportTimeout),
                     new ActionListenerResponseHandler<>(
                         ActionListener.runBefore(fetchRemoteResultListener, () -> Releasables.close(releasable)),
-                        transportActionType.getResponseReader()
+                        transportActionType.getResponseReader(),
+                        ThreadPool.Names.CLUSTER_COORDINATION
                     )
                 );
             }
@@ -1165,7 +1166,7 @@ public void run() {
             public String toString() {
                 return "delayed retrieval of coordination diagnostics info from " + masterEligibleNode;
             }
-        }, remoteRequestInitialDelay, ThreadPool.Names.SAME);
+        }, remoteRequestInitialDelay, ThreadPool.Names.CLUSTER_COORDINATION);
     }
 
     void cancelPollingRemoteMasterStabilityDiagnostic() {

server/src/main/java/org/elasticsearch/discovery/PeerFinder.java

@@ -19,7 +19,6 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
-import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
@@ -107,7 +106,7 @@ public PeerFinder(
 
         transportService.registerRequestHandler(
             REQUEST_PEERS_ACTION_NAME,
-            Names.GENERIC,
+            Names.CLUSTER_COORDINATION,
             false,
             false,
             PeersRequest::new,
@@ -287,20 +286,9 @@ private boolean handleWakeUp() {
             }
         });
 
-        transportService.getThreadPool().scheduleUnlessShuttingDown(findPeersInterval, Names.GENERIC, new AbstractRunnable() {
+        transportService.getThreadPool().scheduleUnlessShuttingDown(findPeersInterval, Names.CLUSTER_COORDINATION, new Runnable() {
             @Override
-            public boolean isForceExecution() {
-                return true;
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                assert false : e;
-                logger.debug("unexpected exception in wakeup", e);
-            }
-
-            @Override
-            protected void doRun() {
+            public void run() {
                 synchronized (mutex) {
                     if (handleWakeUp() == false) {
                         return;
@@ -469,7 +457,7 @@ private void requestPeers() {
 
             final List<DiscoveryNode> knownNodes = List.copyOf(getFoundPeersUnderLock());
 
-            final TransportResponseHandler<PeersResponse> peersResponseHandler = new TransportResponseHandler<PeersResponse>() {
+            final TransportResponseHandler<PeersResponse> peersResponseHandler = new TransportResponseHandler<>() {
 
                 @Override
                 public PeersResponse read(StreamInput in) throws IOException {
@@ -507,7 +495,7 @@ public void handleException(TransportException exp) {
 
                 @Override
                 public String executor() {
-                    return Names.GENERIC;
+                    return Names.CLUSTER_COORDINATION;
                 }
             };
             transportService.sendRequest(

server/src/main/java/org/elasticsearch/discovery/SeedHostsResolver.java

@@ -179,7 +179,7 @@ public void resolveConfiguredHosts(Consumer<List<TransportAddress>> consumer) {
         }
 
         if (resolveInProgress.compareAndSet(false, true)) {
-            transportService.getThreadPool().generic().execute(new AbstractRunnable() {
+            transportService.getThreadPool().executor(ThreadPool.Names.CLUSTER_COORDINATION).execute(new AbstractRunnable() {
                 @Override
                 public void onFailure(Exception e) {
                     logger.debug("failure when resolving unicast hosts list", e);