Fix EsAbortPolicy to conform to API

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

@@ -616,7 +616,7 @@ public static ElasticsearchException[] guessRootCauses(Throwable t) {
         if (ex instanceof ElasticsearchException) {
             return ((ElasticsearchException) ex).guessRootCauses();
         }
-        return new ElasticsearchException[]{new ElasticsearchException(t.getMessage(), t) {
+        return new ElasticsearchException[]{new ElasticsearchException(ex.getMessage(), ex) {
             @Override
             protected String getExceptionName() {
                 return getExceptionName(getCause());
@@ -827,8 +827,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;
 
@@ -831,6 +832,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,6 +853,10 @@ public void writeException(Throwable throwable) throws IOException {
                 writeCause = false;
             } else if (throwable instanceof IOException) {
                 writeVInt(17);
+            } else if (throwable instanceof EsRejectedExecutionException) {
+                writeVInt(18);
+                writeBoolean(((EsRejectedExecutionException) throwable).isExecutorShutdown());
+                writeCause = false;
             } else {
                 ElasticsearchException ex;
                 if (throwable instanceof ElasticsearchException && ElasticsearchException.isRegistered(throwable.getClass(), version)) {

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;
@@ -161,6 +163,16 @@ public void testGuessRootCause() {
             assertEquals(foobars[0].getCause().getClass(), IllegalArgumentException.class);
             assertEquals(foobars[0].getExceptionName(), "illegal_argument_exception");
         }
+
+        {
+            final EsRejectedExecutionException esRejectedExecutionException = new EsRejectedExecutionException("rejected", randomBoolean());
+            final RemoteTransportException remoteTransportException =
+                    new RemoteTransportException("node", buildNewFakeTransportAddress(), "action", esRejectedExecutionException);
+            final ElasticsearchException[] rootCauses = ElasticsearchException.guessRootCauses(remoteTransportException);
+            assertThat(rootCauses, arrayWithSize(1));
+            assertThat(rootCauses[0].getExceptionName(), equalTo("es_rejected_execution_exception"));
+            assertThat(rootCauses[0].getMessage(), equalTo("rejected"));
+        }
     }
 
     public void testDeduplicate() throws IOException {

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));
+    }
+
 }