Carry on rest status if exceptions are not serializable

Today we loose the RestStatus code for non-serializable exceptions.
This can be tricky if they are supposed to signal certain situations
like authentication errors etc. This commit adds support for carrying on
the exceptions in the NotSerializableExceptoinWrapper

core/src/main/java/org/elasticsearch/common/io/stream/NotSerializableExceptionWrapper.java

@@ -20,7 +20,9 @@
 package org.elasticsearch.common.io.stream;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.rest.RestStatus;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -37,10 +39,12 @@ import java.util.Map;
 public final class NotSerializableExceptionWrapper extends ElasticsearchException.WithRestHeadersException {
 
     private final String name;
+    private final RestStatus status;
 
     public NotSerializableExceptionWrapper(Throwable other, Map<String, List<String>> headers) {
         super(other.getMessage(), other.getCause(), headers);
         this.name = ElasticsearchException.getExceptionName(other);
+        this.status = ExceptionsHelper.status(other);
         setStackTrace(other.getStackTrace());
         for (Throwable otherSuppressed : other.getSuppressed()) {
             addSuppressed(otherSuppressed);
@@ -58,16 +62,23 @@ public final class NotSerializableExceptionWrapper extends ElasticsearchExceptio
     public NotSerializableExceptionWrapper(StreamInput in) throws IOException {
         super(in);
         name = in.readString();
+        status = RestStatus.readFrom(in);
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeString(name);
+        RestStatus.writeTo(out, status);
     }
 
     @Override
     protected String getExceptionName() {
         return name;
     }
+
+    @Override
+    public RestStatus status() {
+        return status;
+    }
 }

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

@@ -55,6 +55,7 @@ import org.elasticsearch.indices.InvalidIndexTemplateException;
 import org.elasticsearch.indices.recovery.RecoverFilesRecoveryException;
 import org.elasticsearch.percolator.PercolateException;
 import org.elasticsearch.repositories.RepositoryException;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.admin.indices.alias.delete.AliasesMissingException;
 import org.elasticsearch.search.SearchContextMissingException;
 import org.elasticsearch.search.SearchException;
@@ -588,18 +589,30 @@ public class ExceptionSerializationTests extends ElasticsearchTestCase {
         assertEquals("foo", ex.getHeaders().get("foo").get(0));
         assertEquals("bar", ex.getHeaders().get("foo").get(1));
 
+        RestStatus status = randomFrom(RestStatus.values());
         // ensure we are carrying over the headers even if not serialized
-        ElasticsearchException e = serialize((ElasticsearchException)new UnknownHeaderException("msg", new Tuple("foo", new String[]{"foo", "bar"})));
-        assertTrue(e instanceof NotSerializableExceptionWrapper);
-        assertEquals("msg", ex.getMessage());
-        assertEquals(2, ex.getHeaders().get("foo").size());
-        assertEquals("foo", ex.getHeaders().get("foo").get(0));
-        assertEquals("bar", ex.getHeaders().get("foo").get(1));
+        ElasticsearchException serialize = serialize((ElasticsearchException) new UnknownHeaderException("msg", status, new Tuple("foo", new String[]{"foo", "bar"})));
+        assertTrue(serialize instanceof NotSerializableExceptionWrapper);
+        NotSerializableExceptionWrapper e = (NotSerializableExceptionWrapper) serialize;
+        assertEquals("msg", e.getMessage());
+        assertEquals(2, e.getHeaders().get("foo").size());
+        assertEquals("foo", e.getHeaders().get("foo").get(0));
+        assertEquals("bar", e.getHeaders().get("foo").get(1));
+        assertSame(status, e.status());
+
     }
 
     public static class UnknownHeaderException extends ElasticsearchException.WithRestHeadersException {
-        public UnknownHeaderException(String msg, Tuple<String, String[]>... headers) {
+        private final RestStatus status;
+
+        public UnknownHeaderException(String msg, RestStatus status, Tuple<String, String[]>... headers) {
             super(msg, headers);
+            this.status = status;
+        }
+
+        @Override
+        public RestStatus status() {
+            return status;
         }
     }
 }