Return 429 instead of 500 for timeout handlers (#133111)

This is consistent with the existing ElasticsearchTimeoutException.

Resolves: ES-12238

Author: ywangd

docs/changelog/133111.yaml

@@ -0,0 +1,5 @@
+pr: 133111
+summary: Return 429 instead of 500 for timeout handlers
+area: Network
+type: enhancement
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java

@@ -8,6 +8,7 @@
  */
 package org.elasticsearch.action.admin.cluster.node.tasks;
 
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
@@ -41,6 +42,7 @@
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.tasks.RemovedTaskListener;
 import org.elasticsearch.tasks.Task;
@@ -677,9 +679,14 @@ private void waitForTimeoutTestCase(Function<TaskId, ? extends Iterable<? extend
             Iterable<? extends Throwable> failures = wait.apply(taskId);
 
             for (Throwable failure : failures) {
-                assertNotNull(
-                    ExceptionsHelper.unwrap(failure, ElasticsearchTimeoutException.class, ReceiveTimeoutTransportException.class)
+                final Throwable cause = ExceptionsHelper.unwrap(
+                    failure,
+                    ElasticsearchTimeoutException.class,
+                    ReceiveTimeoutTransportException.class
                 );
+                assertNotNull(cause);
+                assertThat(asInstanceOf(ElasticsearchException.class, cause).status(), equalTo(RestStatus.TOO_MANY_REQUESTS));
+                assertTrue(asInstanceOf(ElasticsearchException.class, cause).isTimeout());
             }
         } finally {
             // Now we can unblock those requests

server/src/main/java/org/elasticsearch/transport/ReceiveTimeoutTransportException.java

@@ -11,6 +11,7 @@
 
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.rest.RestStatus;
 
 import java.io.IOException;
 
@@ -29,4 +30,15 @@ public synchronized Throwable fillInStackTrace() {
         // stack trace is uninformative
         return this;
     }
+
+    @Override
+    public RestStatus status() {
+        // closest thing to "your request took longer than you asked for"
+        return RestStatus.TOO_MANY_REQUESTS;
+    }
+
+    @Override
+    public boolean isTimeout() {
+        return true;
+    }
 }