Remove trappy timeout from ClusterSearchShardsRequest (elastic#111442)

Exposes the `?master_timeout` parameter to the REST API and sets it
appropriately on internal/test requests.

Relates elastic#107984

Author: DaveCTurner

docs/reference/search/search-shards.asciidoc

@@ -63,6 +63,7 @@ include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=preference]
 
 include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=routing]
 
+include::{es-ref-dir}/rest-api/common-parms.asciidoc[tag=master-timeout]
 
 [[search-shards-api-example]]
 ==== {api-examples-title}

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java

@@ -585,7 +585,9 @@ public void testResolvabilityOfDataStreamsInAPIs() throws Exception {
         verifyResolvability(dataStreamName, indicesAdmin().prepareOpen(dataStreamName), false);
         verifyResolvability(dataStreamName, indicesAdmin().prepareClose(dataStreamName), true);
         verifyResolvability(aliasToDataStream, indicesAdmin().prepareClose(aliasToDataStream), true);
-        verifyResolvability(client().execute(TransportClusterSearchShardsAction.TYPE, new ClusterSearchShardsRequest(dataStreamName)));
+        verifyResolvability(
+            client().execute(TransportClusterSearchShardsAction.TYPE, new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, dataStreamName))
+        );
         verifyResolvability(client().execute(TransportIndicesShardStoresAction.TYPE, new IndicesShardStoresRequest(dataStreamName)));
 
         request = new CreateDataStreamAction.Request("logs-barbaz");
@@ -629,7 +631,12 @@ public void testResolvabilityOfDataStreamsInAPIs() throws Exception {
         verifyResolvability(wildcardExpression, indicesAdmin().prepareGetIndex().addIndices(wildcardExpression), false);
         verifyResolvability(wildcardExpression, indicesAdmin().prepareOpen(wildcardExpression), false);
         verifyResolvability(wildcardExpression, indicesAdmin().prepareClose(wildcardExpression), false);
-        verifyResolvability(client().execute(TransportClusterSearchShardsAction.TYPE, new ClusterSearchShardsRequest(wildcardExpression)));
+        verifyResolvability(
+            client().execute(
+                TransportClusterSearchShardsAction.TYPE,
+                new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, wildcardExpression)
+            )
+        );
         verifyResolvability(client().execute(TransportIndicesShardStoresAction.TYPE, new IndicesShardStoresRequest(wildcardExpression)));
     }
 

modules/reindex/src/main/java/org/elasticsearch/reindex/BulkByScrollParallelizationHelper.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsRequest;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsResponse;
+import org.elasticsearch.action.admin.cluster.shards.TransportClusterSearchShardsAction;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -114,12 +115,14 @@ static <Request extends AbstractBulkByScrollRequest<Request>> void initTaskState
     ) {
         int configuredSlices = request.getSlices();
         if (configuredSlices == AbstractBulkByScrollRequest.AUTO_SLICES) {
-            ClusterSearchShardsRequest shardsRequest = new ClusterSearchShardsRequest();
-            shardsRequest.indices(request.getSearchRequest().indices());
-            client.admin().cluster().searchShards(shardsRequest, listener.safeMap(response -> {
-                setWorkerCount(request, task, countSlicesBasedOnShards(response));
-                return null;
-            }));
+            client.execute(
+                TransportClusterSearchShardsAction.TYPE,
+                new ClusterSearchShardsRequest(request.getTimeout(), request.getSearchRequest().indices()),
+                listener.safeMap(response -> {
+                    setWorkerCount(request, task, countSlicesBasedOnShards(response));
+                    return null;
+                })
+            );
         } else {
             setWorkerCount(request, task, configuredSlices);
             listener.onResponse(null);

rest-api-spec/src/main/resources/rest-api-spec/api/search_shards.json

@@ -65,6 +65,10 @@
         ],
         "default":"open",
         "description":"Whether to expand wildcard expression to concrete indices that are open, closed or both."
+      },
+      "master_timeout":{
+        "type":"time",
+        "description":"Explicit operation timeout for connection to master node"
       }
     }
   }

server/src/internalClusterTest/java/org/elasticsearch/action/search/SearchProgressActionListenerIT.java

@@ -202,7 +202,7 @@ private static List<SearchShard> createRandomIndices(Client client) {
         ClusterSearchShardsResponse resp = safeExecute(
             client,
             TransportClusterSearchShardsAction.TYPE,
-            new ClusterSearchShardsRequest("index-*")
+            new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "index-*")
         );
         return Arrays.stream(resp.getGroups()).map(e -> new SearchShard(null, e.getShardId())).sorted().toList();
     }

server/src/internalClusterTest/java/org/elasticsearch/action/termvectors/GetTermVectorsIT.java

@@ -1017,7 +1017,7 @@ public void testArtificialDocWithPreference() throws InterruptedException, IOExc
         // Get search shards
         ClusterSearchShardsResponse searchShardsResponse = safeExecute(
             TransportClusterSearchShardsAction.TYPE,
-            new ClusterSearchShardsRequest("test")
+            new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test")
         );
         List<Integer> shardIds = Arrays.stream(searchShardsResponse.getGroups()).map(s -> s.getShardId().id()).toList();
 

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/ShardRoutingRoleIT.java

@@ -550,7 +550,7 @@ public void testSearchRouting() throws Exception {
             }
             // search-shards API
             for (int i = 0; i < 10; i++) {
-                final var search = new ClusterSearchShardsRequest(INDEX_NAME);
+                final var search = new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, INDEX_NAME);
                 switch (randomIntBetween(0, 2)) {
                     case 0 -> search.routing(randomAlphaOfLength(10));
                     case 1 -> search.routing(randomSearchPreference(routingTableWatcher.numShards, internalCluster().getNodeNames()));

server/src/internalClusterTest/java/org/elasticsearch/cluster/shards/ClusterSearchShardsIT.java

@@ -45,15 +45,15 @@ protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
     public void testSingleShardAllocation() {
         indicesAdmin().prepareCreate("test").setSettings(indexSettings(1, 0).put("index.routing.allocation.include.tag", "A")).get();
         ensureGreen();
-        ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest("test"));
+        ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test"));
         assertThat(response.getGroups().length, equalTo(1));
         assertThat(response.getGroups()[0].getShardId().getIndexName(), equalTo("test"));
         assertThat(response.getGroups()[0].getShardId().getId(), equalTo(0));
         assertThat(response.getGroups()[0].getShards().length, equalTo(1));
         assertThat(response.getNodes().length, equalTo(1));
         assertThat(response.getGroups()[0].getShards()[0].currentNodeId(), equalTo(response.getNodes()[0].getId()));
 
-        response = safeExecute(new ClusterSearchShardsRequest("test").routing("A"));
+        response = safeExecute(new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test").routing("A"));
         assertThat(response.getGroups().length, equalTo(1));
         assertThat(response.getGroups()[0].getShardId().getIndexName(), equalTo("test"));
         assertThat(response.getGroups()[0].getShardId().getId(), equalTo(0));
@@ -67,16 +67,16 @@ public void testMultipleShardsSingleNodeAllocation() {
         indicesAdmin().prepareCreate("test").setSettings(indexSettings(4, 0).put("index.routing.allocation.include.tag", "A")).get();
         ensureGreen();
 
-        ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest("test"));
+        ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test"));
         assertThat(response.getGroups().length, equalTo(4));
         assertThat(response.getGroups()[0].getShardId().getIndexName(), equalTo("test"));
         assertThat(response.getNodes().length, equalTo(1));
         assertThat(response.getGroups()[0].getShards()[0].currentNodeId(), equalTo(response.getNodes()[0].getId()));
 
-        response = safeExecute(new ClusterSearchShardsRequest("test").routing("ABC"));
+        response = safeExecute(new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test").routing("ABC"));
         assertThat(response.getGroups().length, equalTo(1));
 
-        response = safeExecute(new ClusterSearchShardsRequest("test").preference("_shards:2"));
+        response = safeExecute(new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test").preference("_shards:2"));
         assertThat(response.getGroups().length, equalTo(1));
         assertThat(response.getGroups()[0].getShardId().getId(), equalTo(2));
     }
@@ -90,7 +90,7 @@ public void testMultipleIndicesAllocation() {
             .get();
         clusterAdmin().prepareHealth().setWaitForEvents(Priority.LANGUID).setWaitForGreenStatus().get();
 
-        ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest("routing_alias"));
+        ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "routing_alias"));
         assertThat(response.getGroups().length, equalTo(2));
         assertThat(response.getGroups()[0].getShards().length, equalTo(2));
         assertThat(response.getGroups()[1].getShards().length, equalTo(2));
@@ -132,7 +132,7 @@ public void testClusterSearchShardsWithBlocks() {
         )) {
             try {
                 enableIndexBlock("test-blocks", blockSetting);
-                ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest("test-blocks"));
+                ClusterSearchShardsResponse response = safeExecute(new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test-blocks"));
                 assertThat(response.getGroups().length, equalTo(numShards.numPrimaries));
             } finally {
                 disableIndexBlock("test-blocks", blockSetting);
@@ -149,7 +149,11 @@ public void testClusterSearchShardsWithBlocks() {
                     ExceptionsHelper.unwrapCause(
                         safeAwaitFailure(
                             ClusterSearchShardsResponse.class,
-                            l -> client().execute(TransportClusterSearchShardsAction.TYPE, new ClusterSearchShardsRequest("test-blocks"), l)
+                            l -> client().execute(
+                                TransportClusterSearchShardsAction.TYPE,
+                                new ClusterSearchShardsRequest(TEST_REQUEST_TIMEOUT, "test-blocks"),
+                                l
+                            )
                         )
                     )
                 )

server/src/main/java/org/elasticsearch/action/admin/cluster/shards/ClusterSearchShardsRequest.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.TimeValue;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -31,22 +32,16 @@ public final class ClusterSearchShardsRequest extends MasterNodeReadRequest<Clus
     private String preference;
     private IndicesOptions indicesOptions = IndicesOptions.lenientExpandOpen();
 
-    public ClusterSearchShardsRequest() {
-        super(TRAPPY_IMPLICIT_DEFAULT_MASTER_NODE_TIMEOUT);
-    }
-
-    public ClusterSearchShardsRequest(String... indices) {
-        super(TRAPPY_IMPLICIT_DEFAULT_MASTER_NODE_TIMEOUT);
+    public ClusterSearchShardsRequest(TimeValue masterNodeTimeout, String... indices) {
+        super(masterNodeTimeout);
         indices(indices);
     }
 
     public ClusterSearchShardsRequest(StreamInput in) throws IOException {
         super(in);
         indices = in.readStringArray();
-
         routing = in.readOptionalString();
         preference = in.readOptionalString();
-
         indicesOptions = IndicesOptions.readIndicesOptions(in);
     }
 
@@ -56,7 +51,6 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeStringArray(indices);
         out.writeOptionalString(routing);
         out.writeOptionalString(preference);
-
         indicesOptions.writeIndicesOptions(out);
     }