Fix EsAbortPolicy to conform to API (#29075)

The rejected execution handler API says that rejectedExecution(Runnable,
ThreadPoolExecutor) throws a RejectedExecutionException if the task must
be rejected due to capacity on the executor. We do throw something that
smells like a RejectedExecutionException (it is named
EsRejectedExecutionException) yet we violate the API because
EsRejectedExecutionException is not a RejectedExecutionException. This
has caused problems before where we try to catch RejectedExecution when
invoking rejectedExecution but this causes EsRejectedExecutionException
to go uncaught. This commit addresses this by modifying
EsRejectedExecutionException to extend
RejectedExecutionException.

Author: jasontedor

modules/reindex/src/test/java/org/elasticsearch/index/reindex/AsyncBulkByScrollActionTests.java

@@ -115,6 +115,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.hasToString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
@@ -330,7 +331,8 @@ public ScheduledFuture<?> schedule(TimeValue delay, String name, Runnable comman
         ScrollableHitSource.Response response = new ScrollableHitSource.Response(false, emptyList(), 0, emptyList(), null);
         simulateScrollResponse(new DummyAsyncBulkByScrollAction(), timeValueNanos(System.nanoTime()), 10, response);
         ExecutionException e = expectThrows(ExecutionException.class, () -> listener.get());
-        assertThat(e.getMessage(), equalTo("EsRejectedExecutionException[test]"));
+        assertThat(e.getCause(), instanceOf(EsRejectedExecutionException.class));
+        assertThat(e.getCause(), hasToString(containsString("test")));
         assertThat(client.scrollsCleared, contains(scrollId));
 
         // When the task is rejected we don't increment the throttled timer

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -828,8 +828,7 @@ private enum ElasticsearchExceptionHandle {
                 org.elasticsearch.indices.IndexTemplateMissingException::new, 57, UNKNOWN_VERSION_ADDED),
         SEND_REQUEST_TRANSPORT_EXCEPTION(org.elasticsearch.transport.SendRequestTransportException.class,
                 org.elasticsearch.transport.SendRequestTransportException::new, 58, UNKNOWN_VERSION_ADDED),
-        ES_REJECTED_EXECUTION_EXCEPTION(org.elasticsearch.common.util.concurrent.EsRejectedExecutionException.class,
-                org.elasticsearch.common.util.concurrent.EsRejectedExecutionException::new, 59, UNKNOWN_VERSION_ADDED),
+        // 59 used to be EsRejectedExecutionException
         // 60 used to be for EarlyTerminationException
         // 61 used to be for RoutingValidationException
         NOT_SERIALIZABLE_EXCEPTION_WRAPPER(org.elasticsearch.common.io.stream.NotSerializableExceptionWrapper.class,

server/src/main/java/org/elasticsearch/ExceptionsHelper.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.rest.RestStatus;
 
@@ -67,6 +68,8 @@ public static RestStatus status(Throwable t) {
                 return ((ElasticsearchException) t).status();
             } else if (t instanceof IllegalArgumentException) {
                 return RestStatus.BAD_REQUEST;
+            } else if (t instanceof EsRejectedExecutionException) {
+                return RestStatus.TOO_MANY_REQUESTS;
             }
         }
         return RestStatus.INTERNAL_SERVER_ERROR;

server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.geo.GeoPoint;
 import org.elasticsearch.common.text.Text;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
@@ -747,6 +748,11 @@ public <T extends Exception> T readException() throws IOException {
             switch (key) {
                 case 0:
                     final int ord = readVInt();
+                    if (ord == 59) {
+                        final ElasticsearchException ex = new ElasticsearchException(this);
+                        final boolean isExecutorShutdown = readBoolean();
+                        return (T) new EsRejectedExecutionException(ex.getMessage(), isExecutorShutdown);
+                    }
                     return (T) ElasticsearchException.readException(this, ord);
                 case 1:
                     String msg1 = readOptionalString();
@@ -831,6 +837,9 @@ public <T extends Exception> T readException() throws IOException {
                     return (T) readStackTrace(new InterruptedException(readOptionalString()), this);
                 case 17:
                     return (T) readStackTrace(new IOException(readOptionalString(), readException()), this);
+                case 18:
+                    final boolean isExecutorShutdown = readBoolean();
+                    return (T) readStackTrace(new EsRejectedExecutionException(readOptionalString(), isExecutorShutdown), this);
                 default:
                     throw new IOException("no such exception for id: " + key);
             }

server/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.common.geo.GeoPoint;
 import org.elasticsearch.common.io.stream.Writeable.Writer;
 import org.elasticsearch.common.text.Text;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.joda.time.DateTimeZone;
 import org.joda.time.ReadableInstant;
 
@@ -852,8 +853,26 @@ public void writeException(Throwable throwable) throws IOException {
                 writeCause = false;
             } else if (throwable instanceof IOException) {
                 writeVInt(17);
+            } else if (throwable instanceof EsRejectedExecutionException) {
+                if (version.before(Version.V_6_3_0)) {
+                    /*
+                     * This is a backwards compatibility layer when speaking to nodes that still treated EsRejectedExceutionException as an
+                     * instance of ElasticsearchException. As such, we serialize this in a way that the receiving node would read this as an
+                     * EsRejectedExecutionException.
+                     */
+                    final ElasticsearchException ex = new ElasticsearchException(throwable.getMessage());
+                    writeVInt(0);
+                    writeVInt(59);
+                    ex.writeTo(this);
+                    writeBoolean(((EsRejectedExecutionException) throwable).isExecutorShutdown());
+                    return;
+                } else {
+                    writeVInt(18);
+                    writeBoolean(((EsRejectedExecutionException) throwable).isExecutorShutdown());
+                    writeCause = false;
+                }
             } else {
-                ElasticsearchException ex;
+                final ElasticsearchException ex;
                 if (throwable instanceof ElasticsearchException && ElasticsearchException.isRegistered(throwable.getClass(), version)) {
                     ex = (ElasticsearchException) throwable;
                 } else {
@@ -863,7 +882,6 @@ public void writeException(Throwable throwable) throws IOException {
                 writeVInt(ElasticsearchException.getId(ex.getClass()));
                 ex.writeTo(this);
                 return;
-
             }
             if (writeMessage) {
                 writeOptionalString(throwable.getMessage());

server/src/main/java/org/elasticsearch/common/util/concurrent/EsRejectedExecutionException.java

@@ -19,14 +19,9 @@
 
 package org.elasticsearch.common.util.concurrent;
 
-import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.rest.RestStatus;
+import java.util.concurrent.RejectedExecutionException;
 
-import java.io.IOException;
-
-public class EsRejectedExecutionException extends ElasticsearchException {
+public class EsRejectedExecutionException extends RejectedExecutionException {
 
     private final boolean isExecutorShutdown;
 
@@ -43,22 +38,6 @@ public EsRejectedExecutionException() {
         this(null, false);
     }
 
-    @Override
-    public RestStatus status() {
-        return RestStatus.TOO_MANY_REQUESTS;
-    }
-
-    public EsRejectedExecutionException(StreamInput in) throws IOException{
-        super(in);
-        isExecutorShutdown = in.readBoolean();
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        super.writeTo(out);
-        out.writeBoolean(isExecutorShutdown);
-    }
-
     /**
      * Checks if the thread pool that rejected the execution was terminated
      * shortly after the rejection. Its possible that this returns false and the

server/src/test/java/org/elasticsearch/ElasticsearchExceptionTests.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -76,6 +77,7 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertToXContentEquivalent;
 import static org.hamcrest.CoreMatchers.hasItem;
 import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.Matchers.arrayWithSize;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.startsWith;

server/src/test/java/org/elasticsearch/ExceptionSerializationTests.java

@@ -728,7 +728,7 @@ public void testIds() {
         ids.put(56, org.elasticsearch.common.settings.SettingsException.class);
         ids.put(57, org.elasticsearch.indices.IndexTemplateMissingException.class);
         ids.put(58, org.elasticsearch.transport.SendRequestTransportException.class);
-        ids.put(59, org.elasticsearch.common.util.concurrent.EsRejectedExecutionException.class);
+        ids.put(59, null); // weas EsRejectedExecutionException, which is no longer an instance of ElasticsearchException
         ids.put(60, null); // EarlyTerminationException was removed in 6.0
         ids.put(61, null); // RoutingValidationException was removed in 5.0
         ids.put(62, org.elasticsearch.common.io.stream.NotSerializableExceptionWrapper.class);

server/src/test/java/org/elasticsearch/ExceptionsHelperTests.java

@@ -20,6 +20,8 @@
 package org.elasticsearch;
 
 import org.apache.commons.codec.DecoderException;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.Optional;
@@ -84,4 +86,9 @@ private void assertError(final Throwable cause, final Error error) {
         assertThat(maybeError.get(), equalTo(error));
     }
 
+    public void testStatus() {
+        assertThat(ExceptionsHelper.status(new IllegalArgumentException("illegal")), equalTo(RestStatus.BAD_REQUEST));
+        assertThat(ExceptionsHelper.status(new EsRejectedExecutionException("rejected")), equalTo(RestStatus.TOO_MANY_REQUESTS));
+    }
+
 }