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

@@ -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;
 
@@ -747,6 +748,13 @@ public <T extends Exception> T readException() throws IOException {
             switch (key) {
                 case 0:
                     final int ord = readVInt();
+                    // TODO: remove the if branch when master is bumped to 8.0.0
+                    assert Version.CURRENT.major < 8;
+                    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 +839,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,28 @@ public void writeException(Throwable throwable) throws IOException {
                 writeCause = false;
             } else if (throwable instanceof IOException) {
                 writeVInt(17);
+            } else if (throwable instanceof EsRejectedExecutionException) {
+                // TODO: remove the if branch when master is bumped to 8.0.0
+                assert Version.CURRENT.major < 8;
+                if (version.before(Version.V_7_0_0_alpha1)) {
+                    /*
+                     * 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 +884,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));
+    }
+
 }