Stop returning "es." internal exception headers as http response headers (#22703)

move "es." internal headers to separate metadata set in ElasticsearchException and stop returning them as response headers

Closes #17593

* [TEST] remove ESExceptionTests, move its methods to ElasticsearchExceptionTests or ExceptionSerializationTests

13 files changed, 656 insertions, 565 deletions

diff --git a/buildSrc/src/main/resources/checkstyle_suppressions.xml b/buildSrc/src/main/resources/checkstyle_suppressions.xml
index 61e9a205d3..ef2509ff8e 100644
--- a/buildSrc/src/main/resources/checkstyle_suppressions.xml
+++ b/buildSrc/src/main/resources/checkstyle_suppressions.xml
@@ -161,7 +161,6 @@
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]ingest[/\\]SimulatePipelineRequestBuilder.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]ingest[/\\]SimulatePipelineTransportAction.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]search[/\\]MultiSearchRequestBuilder.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]search[/\\]SearchPhaseExecutionException.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]search[/\\]SearchResponse.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]search[/\\]ShardSearchFailure.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]search[/\\]TransportClearScrollAction.java" checks="LineLength" />
@@ -533,7 +532,6 @@
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]threadpool[/\\]ThreadPool.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]apache[/\\]lucene[/\\]queries[/\\]BlendedTermQueryTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]apache[/\\]lucene[/\\]search[/\\]postingshighlight[/\\]CustomPostingsHighlighterTests.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]ESExceptionTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]NamingConventionTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]VersionTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]action[/\\]RejectionActionIT.java" checks="LineLength" />
diff --git a/core/src/main/java/org/elasticsearch/ElasticsearchException.java b/core/src/main/java/org/elasticsearch/ElasticsearchException.java
index c62ffe5e4c..de2b6054e9 100644
--- a/core/src/main/java/org/elasticsearch/ElasticsearchException.java
+++ b/core/src/main/java/org/elasticsearch/ElasticsearchException.java
@@ -37,8 +37,9 @@ import org.elasticsearch.transport.TcpTransport;
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -56,14 +57,14 @@ import static org.elasticsearch.common.xcontent.XContentParserUtils.throwUnknown
  */
 public class ElasticsearchException extends RuntimeException implements ToXContent, Writeable {
 
-    static final Version UNKNOWN_VERSION_ADDED = Version.fromId(0);
+    private static final Version UNKNOWN_VERSION_ADDED = Version.fromId(0);
 
     /**
      * Passed in the {@link Params} of {@link #generateThrowableXContent(XContentBuilder, Params, Throwable)}
      * to control if the {@code caused_by} element should render. Unlike most parameters to {@code toXContent} methods this parameter is
      * internal only and not available as a URL parameter.
      */
-    public static final String REST_EXCEPTION_SKIP_CAUSE = "rest.exception.cause.skip";
+    private static final String REST_EXCEPTION_SKIP_CAUSE = "rest.exception.cause.skip";
     /**
      * Passed in the {@link Params} of {@link #generateThrowableXContent(XContentBuilder, Params, Throwable)}
      * to control if the {@code stack_trace} element should render. Unlike most parameters to {@code toXContent} methods this parameter is
@@ -72,11 +73,11 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     public static final String REST_EXCEPTION_SKIP_STACK_TRACE = "rest.exception.stacktrace.skip";
     public static final boolean REST_EXCEPTION_SKIP_STACK_TRACE_DEFAULT = true;
     private static final boolean REST_EXCEPTION_SKIP_CAUSE_DEFAULT = false;
-    private static final String INDEX_HEADER_KEY = "es.index";
-    private static final String INDEX_HEADER_KEY_UUID = "es.index_uuid";
-    private static final String SHARD_HEADER_KEY = "es.shard";
-    private static final String RESOURCE_HEADER_TYPE_KEY = "es.resource.type";
-    private static final String RESOURCE_HEADER_ID_KEY = "es.resource.id";
+    private static final String INDEX_METADATA_KEY = "es.index";
+    private static final String INDEX_METADATA_KEY_UUID = "es.index_uuid";
+    private static final String SHARD_METADATA_KEY = "es.shard";
+    private static final String RESOURCE_METADATA_TYPE_KEY = "es.resource.type";
+    private static final String RESOURCE_METADATA_ID_KEY = "es.resource.id";
 
     private static final String TYPE = "type";
     private static final String REASON = "reason";
@@ -88,6 +89,7 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
 
     private static final Map<Integer, CheckedFunction<StreamInput, ? extends ElasticsearchException, IOException>> ID_TO_SUPPLIER;
     private static final Map<Class<? extends ElasticsearchException>, ElasticsearchExceptionHandle> CLASS_TO_ELASTICSEARCH_EXCEPTION_HANDLE;
+    private final Map<String, List<String>> metadata = new HashMap<>();
     private final Map<String, List<String>> headers = new HashMap<>();
 
     /**
@@ -129,14 +131,57 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
         super(in.readOptionalString(), in.readException());
         readStackTrace(this, in);
         headers.putAll(in.readMapOfLists(StreamInput::readString, StreamInput::readString));
+        //TODO change to onOrAfter once backported to 5.x
+        if (in.getVersion().after(Version.V_5_3_0_UNRELEASED)) {
+            metadata.putAll(in.readMapOfLists(StreamInput::readString, StreamInput::readString));
+        } else {
+            for (Iterator<Map.Entry<String, List<String>>> iterator = headers.entrySet().iterator(); iterator.hasNext(); ) {
+                Map.Entry<String, List<String>> header = iterator.next();
+                if (header.getKey().startsWith("es.")) {
+                    metadata.put(header.getKey(), header.getValue());
+                    iterator.remove();
+                }
+            }
+        }
     }
 
     /**
-     * Adds a new header with the given key.
-     * This method will replace existing header if a header with the same key already exists
+     * Adds a new piece of metadata with the given key.
+     * If the provided key is already present, the corresponding metadata will be replaced
      */
-    public void addHeader(String key, String... value) {
-        this.headers.put(key, Arrays.asList(value));
+    public void addMetadata(String key, String... values) {
+        addMetadata(key, Arrays.asList(values));
+    }
+
+    /**
+     * Adds a new piece of metadata with the given key.
+     * If the provided key is already present, the corresponding metadata will be replaced
+     */
+    public void addMetadata(String key, List<String> values) {
+        //we need to enforce this otherwise bw comp doesn't work properly, as "es." was the previous criteria to split headers in two sets
+        if (key.startsWith("es.") == false) {
+            throw new IllegalArgumentException("exception metadata must start with [es.], found [" + key + "] instead");
+        }
+        this.metadata.put(key, values);
+    }
+
+    /**
+     * Returns a set of all metadata keys on this exception
+     */
+    public Set<String> getMetadataKeys() {
+        return metadata.keySet();
+    }
+
+    /**
+     * Returns the list of metadata values for the given key or {@code null} if no metadata for the
+     * given key exists.
+     */
+    public List<String> getMetadata(String key) {
+        return metadata.get(key);
+    }
+
+    protected Map<String, List<String>> getMetadata() {
+        return metadata;
     }
 
     /**
@@ -144,9 +189,20 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
      * This method will replace existing header if a header with the same key already exists
      */
     public void addHeader(String key, List<String> value) {
+        //we need to enforce this otherwise bw comp doesn't work properly, as "es." was the previous criteria to split headers in two sets
+        if (key.startsWith("es.")) {
+            throw new IllegalArgumentException("exception headers must not start with [es.], found [" + key + "] instead");
+        }
         this.headers.put(key, value);
     }
 
+    /**
+     * Adds a new header with the given key.
+     * This method will replace existing header if a header with the same key already exists
+     */
+    public void addHeader(String key, String... value) {
+        addHeader(key, Arrays.asList(value));
+    }
 
     /**
      * Returns a set of all header keys on this exception
@@ -156,7 +212,7 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     }
 
     /**
-     * Returns the list of header values for the given key or {@code null} if not header for the
+     * Returns the list of header values for the given key or {@code null} if no header for the
      * given key exists.
      */
     public List<String> getHeader(String key) {
@@ -227,7 +283,16 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
         out.writeOptionalString(this.getMessage());
         out.writeException(this.getCause());
         writeStackTraces(this, out);
-        out.writeMapOfLists(headers, StreamOutput::writeString, StreamOutput::writeString);
+        //TODO change to onOrAfter once backported to 5.x
+        if (out.getVersion().after(Version.V_5_3_0_UNRELEASED)) {
+            out.writeMapOfLists(headers, StreamOutput::writeString, StreamOutput::writeString);
+            out.writeMapOfLists(metadata, StreamOutput::writeString, StreamOutput::writeString);
+        } else {
+            HashMap<String, List<String>> finalHeaders = new HashMap<>(headers.size() + metadata.size());
+            finalHeaders.putAll(headers);
+            finalHeaders.putAll(metadata);
+            out.writeMapOfLists(finalHeaders, StreamOutput::writeString, StreamOutput::writeString);
+        }
     }
 
     public static ElasticsearchException readException(StreamInput input, int id) throws IOException {
@@ -266,24 +331,19 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
         if (ex != this) {
             generateThrowableXContent(builder, params, this);
         } else {
-            innerToXContent(builder, params, this, getExceptionName(), getMessage(), headers, getCause());
+            innerToXContent(builder, params, this, getExceptionName(), getMessage(), headers, metadata, getCause());
         }
         return builder;
     }
 
     protected static void innerToXContent(XContentBuilder builder, Params params,
                                           Throwable throwable, String type, String message, Map<String, List<String>> headers,
-                                          Throwable cause) throws IOException {
+                                          Map<String, List<String>> metadata, Throwable cause) throws IOException {
         builder.field(TYPE, type);
         builder.field(REASON, message);
 
-        Set<String> customHeaders = new HashSet<>();
-        for (String key : headers.keySet()) {
-            if (key.startsWith("es.")) {
-                headerToXContent(builder, key.substring("es.".length()), headers.get(key));
-            } else {
-                customHeaders.add(key);
-            }
+        for (Map.Entry<String, List<String>> entry : metadata.entrySet()) {
+            headerToXContent(builder, entry.getKey().substring("es.".length()), entry.getValue());
         }
 
         if (throwable instanceof ElasticsearchException) {
@@ -300,10 +360,10 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
             }
         }
 
-        if (customHeaders.isEmpty() == false) {
+        if (headers.isEmpty() == false) {
             builder.startObject(HEADER);
-            for (String header : customHeaders) {
-                headerToXContent(builder, header, headers.get(header));
+            for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
+                headerToXContent(builder, entry.getKey(), entry.getValue());
             }
             builder.endObject();
         }
@@ -336,7 +396,7 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     /**
      * Static toXContent helper method that renders {@link org.elasticsearch.ElasticsearchException} or {@link Throwable} instances
      * as XContent, delegating the rendering to {@link #toXContent(XContentBuilder, Params)}
-     * or {@link #innerToXContent(XContentBuilder, Params, Throwable, String, String, Map, Throwable)}.
+     * or {@link #innerToXContent(XContentBuilder, Params, Throwable, String, String, Map, Map, Throwable)}.
      *
      * This method is usually used when the {@link Throwable} is rendered as a part of another XContent object.
      */
@@ -346,7 +406,7 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
         if (t instanceof ElasticsearchException) {
             ((ElasticsearchException) t).toXContent(builder, params);
         } else {
-            innerToXContent(builder, params, t, getExceptionName(t), t.getMessage(), emptyMap(), t.getCause());
+            innerToXContent(builder, params, t, getExceptionName(t), t.getMessage(), emptyMap(), emptyMap(), t.getCause());
         }
     }
 
@@ -410,6 +470,7 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
 
         String type = null, reason = null, stack = null;
         ElasticsearchException cause = null;
+        Map<String, List<String>> metadata = new HashMap<>();
         Map<String, Object> headers = new HashMap<>();
 
         do {
@@ -423,8 +484,7 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
                 } else if (STACK_TRACE.equals(currentFieldName)) {
                     stack = parser.text();
                 } else {
-                    // Everything else is considered as a header
-                    headers.put(currentFieldName, parser.text());
+                    metadata.put(currentFieldName, Collections.singletonList(parser.text()));
                 }
             } else if (token == XContentParser.Token.START_OBJECT) {
                 if (CAUSED_BY.equals(currentFieldName)) {
@@ -446,6 +506,16 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
         message.append(']');
 
         ElasticsearchException e = new ElasticsearchException(message.toString(), cause);
+
+        for (Map.Entry<String, List<String>> entry : metadata.entrySet()) {
+            //subclasses can print out additional metadata through the metadataToXContent method. Simple key-value pairs will be
+            //parsed back and become part of this metadata set, while objects and arrays are not supported when parsing back.
+            //Those key-value pairs become part of the metadata set and inherit the "es." prefix as that is currently required
+            //by addMetadata. The prefix will get stripped out when printing metadata out so it will be effectively invisible.
+            //TODO move subclasses that print out simple metadata to using addMetadata directly and support also numbers and booleans.
+            //TODO rename metadataToXContent and have only SearchPhaseExecutionException use it, which prints out complex objects
+            e.addMetadata("es." + entry.getKey(), entry.getValue());
+        }
         for (Map.Entry<String, Object> header : headers.entrySet()) {
             e.addHeader(header.getKey(), String.valueOf(header.getValue()));
         }
@@ -500,9 +570,9 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     @Override
     public String toString() {
         StringBuilder builder = new StringBuilder();
-        if (headers.containsKey(INDEX_HEADER_KEY)) {
+        if (metadata.containsKey(INDEX_METADATA_KEY)) {
             builder.append(getIndex());
-            if (headers.containsKey(SHARD_HEADER_KEY)) {
+            if (metadata.containsKey(SHARD_METADATA_KEY)) {
                 builder.append('[').append(getShardId()).append(']');
             }
             builder.append(' ');
@@ -863,9 +933,9 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     }
 
     public Index getIndex() {
-        List<String> index = getHeader(INDEX_HEADER_KEY);
+        List<String> index = getMetadata(INDEX_METADATA_KEY);
         if (index != null && index.isEmpty() == false) {
-            List<String> index_uuid = getHeader(INDEX_HEADER_KEY_UUID);
+            List<String> index_uuid = getMetadata(INDEX_METADATA_KEY_UUID);
             return new Index(index.get(0), index_uuid.get(0));
         }
 
@@ -873,7 +943,7 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     }
 
     public ShardId getShardId() {
-        List<String> shard = getHeader(SHARD_HEADER_KEY);
+        List<String> shard = getMetadata(SHARD_METADATA_KEY);
         if (shard != null && shard.isEmpty() == false) {
             return new ShardId(getIndex(), Integer.parseInt(shard.get(0)));
         }
@@ -882,8 +952,8 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
 
     public void setIndex(Index index) {
         if (index != null) {
-            addHeader(INDEX_HEADER_KEY, index.getName());
-            addHeader(INDEX_HEADER_KEY_UUID, index.getUUID());
+            addMetadata(INDEX_METADATA_KEY, index.getName());
+            addMetadata(INDEX_METADATA_KEY_UUID, index.getUUID());
         }
     }
 
@@ -896,27 +966,22 @@ public class ElasticsearchException extends RuntimeException implements ToXConte
     public void setShard(ShardId shardId) {
         if (shardId != null) {
             setIndex(shardId.getIndex());
-            addHeader(SHARD_HEADER_KEY, Integer.toString(shardId.id()));
+            addMetadata(SHARD_METADATA_KEY, Integer.toString(shardId.id()));
         }
     }
 
-    public void setShard(String index, int shardId) {
-            setIndex(index);
-            addHeader(SHARD_HEADER_KEY, Integer.toString(shardId));
-    }
-
     public void setResources(String type, String... id) {
         assert type != null;
-        addHeader(RESOURCE_HEADER_ID_KEY, id);
-        addHeader(RESOURCE_HEADER_TYPE_KEY, type);
+        addMetadata(RESOURCE_METADATA_ID_KEY, id);
+        addMetadata(RESOURCE_METADATA_TYPE_KEY, type);
     }
 
     public List<String> getResourceId() {
-        return getHeader(RESOURCE_HEADER_ID_KEY);
+        return getMetadata(RESOURCE_METADATA_ID_KEY);
     }
 
     public String getResourceType() {
-        List<String> header = getHeader(RESOURCE_HEADER_TYPE_KEY);
+        List<String> header = getMetadata(RESOURCE_METADATA_TYPE_KEY);
         if (header != null && header.isEmpty() == false) {
             assert header.size() == 1;
             return header.get(0);
diff --git a/core/src/main/java/org/elasticsearch/action/search/SearchPhaseExecutionException.java b/core/src/main/java/org/elasticsearch/action/search/SearchPhaseExecutionException.java
index 6e2cfd6046..c6e0b21dff 100644
--- a/core/src/main/java/org/elasticsearch/action/search/SearchPhaseExecutionException.java
+++ b/core/src/main/java/org/elasticsearch/action/search/SearchPhaseExecutionException.java
@@ -138,7 +138,8 @@ public class SearchPhaseExecutionException extends ElasticsearchException {
         builder.field("grouped", group); // notify that it's grouped
         builder.field("failed_shards");
         builder.startArray();
-        ShardOperationFailedException[] failures = params.paramAsBoolean("group_shard_failures", true) ? ExceptionsHelper.groupBy(shardFailures) : shardFailures;
+        ShardOperationFailedException[] failures = params.paramAsBoolean("group_shard_failures", true) ?
+                ExceptionsHelper.groupBy(shardFailures) : shardFailures;
         for (ShardOperationFailedException failure : failures) {
             builder.startObject();
             failure.toXContent(builder, params);
@@ -156,7 +157,7 @@ public class SearchPhaseExecutionException extends ElasticsearchException {
             // We don't have a cause when all shards failed, but we do have shards failures so we can "guess" a cause
             // (see {@link #getCause()}). Here, we use super.getCause() because we don't want the guessed exception to
             // be rendered twice (one in the "cause" field, one in "failed_shards")
-            innerToXContent(builder, params, this, getExceptionName(), getMessage(), getHeaders(), super.getCause());
+            innerToXContent(builder, params, this, getExceptionName(), getMessage(), getHeaders(), getMetadata(), super.getCause());
         }
         return builder;
     }
diff --git a/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java b/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java
index ceca57e052..a9cfaafd2c 100644
--- a/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java
+++ b/core/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java
@@ -438,7 +438,6 @@ public abstract class TransportBroadcastByNodeAction<Request extends BroadcastRe
             } catch (Exception e) {
                 BroadcastShardOperationFailedException failure =
                     new BroadcastShardOperationFailedException(shardRouting.shardId(), "operation " + actionName + " failed", e);
-                failure.setIndex(shardRouting.getIndexName());
                 failure.setShard(shardRouting.shardId());
                 shardResults[shardIndex] = failure;
                 if (TransportActions.isShardNotAvailableException(e)) {
diff --git a/core/src/main/java/org/elasticsearch/common/io/stream/NotSerializableExceptionWrapper.java b/core/src/main/java/org/elasticsearch/common/io/stream/NotSerializableExceptionWrapper.java
index b25d4bc9b7..fd4a215eab 100644
--- a/core/src/main/java/org/elasticsearch/common/io/stream/NotSerializableExceptionWrapper.java
+++ b/core/src/main/java/org/elasticsearch/common/io/stream/NotSerializableExceptionWrapper.java
@@ -38,8 +38,7 @@ public final class NotSerializableExceptionWrapper extends ElasticsearchExceptio
     private final RestStatus status;
 
     public NotSerializableExceptionWrapper(Throwable other) {
-        super(ElasticsearchException.getExceptionName(other) +
-                        ": " + other.getMessage(), other.getCause());
+        super(ElasticsearchException.getExceptionName(other) + ": " + other.getMessage(), other.getCause());
         this.name = ElasticsearchException.getExceptionName(other);
         this.status = ExceptionsHelper.status(other);
         setStackTrace(other.getStackTrace());
@@ -51,6 +50,9 @@ public final class NotSerializableExceptionWrapper extends ElasticsearchExceptio
             for (String key : ex.getHeaderKeys()) {
                 this.addHeader(key, ex.getHeader(key));
             }
+            for (String key : ex.getMetadataKeys()) {
+                this.addMetadata(key, ex.getMetadata(key));
+            }
         }
     }
 
diff --git a/core/src/test/java/org/elasticsearch/ESExceptionTests.java b/core/src/test/java/org/elasticsearch/ESExceptionTests.java
deleted file mode 100644
index 5f09cfe57c..0000000000
--- a/core/src/test/java/org/elasticsearch/ESExceptionTests.java
+++ /dev/null
@@ -1,380 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch licenses this file to you under
- * the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.elasticsearch;
-
-import org.apache.lucene.index.CorruptIndexException;
-import org.apache.lucene.index.IndexFormatTooNewException;
-import org.apache.lucene.index.IndexFormatTooOldException;
-import org.apache.lucene.store.AlreadyClosedException;
-import org.apache.lucene.store.LockObtainFailedException;
-import org.elasticsearch.action.search.SearchPhaseExecutionException;
-import org.elasticsearch.action.search.ShardSearchFailure;
-import org.elasticsearch.common.ParsingException;
-import org.elasticsearch.common.io.stream.BytesStreamOutput;
-import org.elasticsearch.common.io.stream.NotSerializableExceptionWrapper;
-import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.xcontent.ToXContent;
-import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.common.xcontent.XContentFactory;
-import org.elasticsearch.common.xcontent.XContentLocation;
-import org.elasticsearch.index.Index;
-import org.elasticsearch.index.IndexNotFoundException;
-import org.elasticsearch.index.query.QueryShardException;
-import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.search.SearchParseException;
-import org.elasticsearch.search.SearchShardTarget;
-import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.test.TestSearchContext;
-import org.elasticsearch.test.VersionUtils;
-import org.elasticsearch.test.hamcrest.ElasticsearchAssertions;
-import org.elasticsearch.transport.RemoteTransportException;
-import org.hamcrest.Matchers;
-
-import java.io.EOFException;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.nio.file.NoSuchFileException;
-
-import static org.hamcrest.Matchers.equalTo;
-
-public class ESExceptionTests extends ESTestCase {
-    private static final ToXContent.Params PARAMS = ToXContent.EMPTY_PARAMS;
-
-    private class UnknownException extends Exception {
-
-        UnknownException(final String message, final Exception cause) {
-            super(message, cause);
-        }
-
-    }
-
-    public void testStatus() {
-        ElasticsearchException exception = new ElasticsearchException("test");
-        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
-
-        exception = new ElasticsearchException("test", new RuntimeException());
-        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
-
-        exception = new ElasticsearchException("test", new ResourceNotFoundException("test"));
-        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
-
-        exception = new RemoteTransportException("test", new ResourceNotFoundException("test"));
-        assertThat(exception.status(), equalTo(RestStatus.NOT_FOUND));
-
-        exception = new RemoteTransportException("test", new ResourceAlreadyExistsException("test"));
-        assertThat(exception.status(), equalTo(RestStatus.BAD_REQUEST));
-
-        exception = new RemoteTransportException("test", new IllegalArgumentException("foobar"));
-        assertThat(exception.status(), equalTo(RestStatus.BAD_REQUEST));
-
-        exception = new RemoteTransportException("test", new IllegalStateException("foobar"));
-        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
-    }
-
-    public void testGuessRootCause() {
-        {
-            ElasticsearchException exception = new ElasticsearchException("foo", new ElasticsearchException("bar", new IndexNotFoundException("foo", new RuntimeException("foobar"))));
-            ElasticsearchException[] rootCauses = exception.guessRootCauses();
-            assertEquals(rootCauses.length, 1);
-            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "index_not_found_exception");
-            assertEquals(rootCauses[0].getMessage(), "no such index");
-            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
-            ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
-            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed", new ShardSearchFailure[]{failure, failure1});
-            if (randomBoolean()) {
-                rootCauses = (randomBoolean() ? new RemoteTransportException("remoteboom", ex) : ex).guessRootCauses();
-            } else {
-                rootCauses = ElasticsearchException.guessRootCauses(randomBoolean() ? new RemoteTransportException("remoteboom", ex) : ex);
-            }
-            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "parsing_exception");
-            assertEquals(rootCauses[0].getMessage(), "foobar");
-
-            ElasticsearchException oneLevel = new ElasticsearchException("foo", new RuntimeException("foobar"));
-            rootCauses = oneLevel.guessRootCauses();
-            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "exception");
-            assertEquals(rootCauses[0].getMessage(), "foo");
-        }
-        {
-            ShardSearchFailure failure = new ShardSearchFailure(
-                    new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
-            ShardSearchFailure failure1 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 1));
-            ShardSearchFailure failure2 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 2));
-            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed", new ShardSearchFailure[]{failure, failure1, failure2});
-            final ElasticsearchException[] rootCauses = ex.guessRootCauses();
-            assertEquals(rootCauses.length, 2);
-            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "parsing_exception");
-            assertEquals(rootCauses[0].getMessage(), "foobar");
-            assertEquals(((ParsingException) rootCauses[0]).getLineNumber(), 1);
-            assertEquals(((ParsingException) rootCauses[0]).getColumnNumber(), 2);
-            assertEquals(ElasticsearchException.getExceptionName(rootCauses[1]), "query_shard_exception");
-            assertEquals((rootCauses[1]).getIndex().getName(), "foo1");
-            assertEquals(rootCauses[1].getMessage(), "foobar");
-        }
-
-        {
-            final ElasticsearchException[] foobars = ElasticsearchException.guessRootCauses(new IllegalArgumentException("foobar"));
-            assertEquals(foobars.length, 1);
-            assertTrue(foobars[0] instanceof ElasticsearchException);
-            assertEquals(foobars[0].getMessage(), "foobar");
-            assertEquals(foobars[0].getCause().getClass(), IllegalArgumentException.class);
-            assertEquals(foobars[0].getExceptionName(), "illegal_argument_exception");
-        }
-
-    }
-
-    public void testDeduplicate() throws IOException {
-        {
-            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
-            ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
-            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed", randomBoolean() ? failure1.getCause() : failure.getCause(), new ShardSearchFailure[]{failure, failure1});
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ex.toXContent(builder, PARAMS);
-            builder.endObject();
-            String expected = "{\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\",\"phase\":\"search\",\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\",\"reason\":{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}}]}";
-            assertEquals(expected, builder.string());
-        }
-        {
-            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
-            ShardSearchFailure failure1 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 1));
-            ShardSearchFailure failure2 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 2));
-            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed", new ShardSearchFailure[]{failure, failure1, failure2});
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ex.toXContent(builder, PARAMS);
-            builder.endObject();
-            String expected = "{\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\",\"phase\":\"search\",\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\",\"reason\":{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}},{\"shard\":1,\"index\":\"foo1\",\"node\":\"node_1\",\"reason\":{\"type\":\"query_shard_exception\",\"reason\":\"foobar\",\"index_uuid\":\"_na_\",\"index\":\"foo1\"}}]}";
-            assertEquals(expected, builder.string());
-        }
-        {
-            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
-            ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
-                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
-            NullPointerException nullPointerException = new NullPointerException();
-            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed", nullPointerException, new ShardSearchFailure[]{failure, failure1});
-            assertEquals(nullPointerException, ex.getCause());
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ex.toXContent(builder, PARAMS);
-            builder.endObject();
-            String expected = "{\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\",\"phase\":\"search\",\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\",\"reason\":{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}}],\"caused_by\":{\"type\":\"null_pointer_exception\",\"reason\":null}}";
-            assertEquals(expected, builder.string());
-        }
-    }
-
-    /**
-     * Check whether this exception contains an exception of the given type:
-     * either it is of the given class itself or it contains a nested cause
-     * of the given type.
-     *
-     * @param exType the exception type to look for
-     * @return whether there is a nested exception of the specified type
-     */
-    private boolean contains(Throwable t, Class<? extends Throwable> exType) {
-        if (exType == null) {
-            return false;
-        }
-        for (Throwable cause = t; t != null; t = t.getCause()) {
-            if (exType.isInstance(cause)) {
-                return true;
-            }
-        }
-        return false;
-    }
-
-    public void testGetRootCause() {
-        Exception root = new RuntimeException("foobar");
-        ElasticsearchException exception = new ElasticsearchException("foo", new ElasticsearchException("bar", new IllegalArgumentException("index is closed", root)));
-        assertEquals(root, exception.getRootCause());
-        assertTrue(contains(exception, RuntimeException.class));
-        assertFalse(contains(exception, EOFException.class));
-    }
-
-    public void testToString() {
-        ElasticsearchException exception = new ElasticsearchException("foo", new ElasticsearchException("bar", new IllegalArgumentException("index is closed", new RuntimeException("foobar"))));
-        assertEquals("ElasticsearchException[foo]; nested: ElasticsearchException[bar]; nested: IllegalArgumentException[index is closed]; nested: RuntimeException[foobar];", exception.toString());
-    }
-
-    public void testToXContent() throws IOException {
-        {
-            ElasticsearchException ex = new SearchParseException(new TestSearchContext(null), "foo", new XContentLocation(1,0));
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ex.toXContent(builder, PARAMS);
-            builder.endObject();
-
-            String expected = "{\"type\":\"search_parse_exception\",\"reason\":\"foo\",\"line\":1,\"col\":0}";
-            assertEquals(expected, builder.string());
-        }
-        {
-            ElasticsearchException ex = new ElasticsearchException("foo", new ElasticsearchException("bar", new IllegalArgumentException("index is closed", new RuntimeException("foobar"))));
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ex.toXContent(builder, PARAMS);
-            builder.endObject();
-
-            String expected = "{\"type\":\"exception\",\"reason\":\"foo\",\"caused_by\":{\"type\":\"exception\",\"reason\":\"bar\",\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"index is closed\",\"caused_by\":{\"type\":\"runtime_exception\",\"reason\":\"foobar\"}}}}";
-            assertEquals(expected, builder.string());
-        }
-
-        {
-            Exception ex = new FileNotFoundException("foo not found");
-            if (randomBoolean()) {
-                // just a wrapper which is omitted
-                ex = new RemoteTransportException("foobar", ex);
-            }
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ElasticsearchException.generateThrowableXContent(builder, PARAMS, ex);
-            builder.endObject();
-
-            String expected = "{\"type\":\"file_not_found_exception\",\"reason\":\"foo not found\"}";
-            assertEquals(expected, builder.string());
-        }
-
-        {
-            ParsingException ex = new ParsingException(1, 2, "foobar", null);
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ElasticsearchException.generateThrowableXContent(builder, PARAMS, ex);
-            builder.endObject();
-            String expected = "{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}";
-            assertEquals(expected, builder.string());
-        }
-
-        { // test equivalence
-            ElasticsearchException ex =  new RemoteTransportException("foobar", new FileNotFoundException("foo not found"));
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ElasticsearchException.generateThrowableXContent(builder, PARAMS, ex);
-            builder.endObject();
-
-            XContentBuilder otherBuilder = XContentFactory.jsonBuilder();
-
-            otherBuilder.startObject();
-            ex.toXContent(otherBuilder, PARAMS);
-            otherBuilder.endObject();
-            assertEquals(otherBuilder.string(), builder.string());
-            assertEquals("{\"type\":\"file_not_found_exception\",\"reason\":\"foo not found\"}", builder.string());
-        }
-
-        { // render header
-            ParsingException ex = new ParsingException(1, 2, "foobar", null);
-            ex.addHeader("test", "some value");
-            ex.addHeader("test_multi", "some value", "another value");
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            ElasticsearchException.generateThrowableXContent(builder, PARAMS, ex);
-            builder.endObject();
-            assertThat(builder.string(), Matchers.anyOf( // iteration order depends on platform
-                            equalTo("{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2,\"header\":{\"test_multi\":[\"some value\",\"another value\"],\"test\":\"some value\"}}"),
-                            equalTo("{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2,\"header\":{\"test\":\"some value\",\"test_multi\":[\"some value\",\"another value\"]}}")
-            ));
-        }
-    }
-
-    public void testSerializeElasticsearchException() throws IOException {
-        BytesStreamOutput out = new BytesStreamOutput();
-        ParsingException ex = new ParsingException(1, 2, "foobar", null);
-        out.writeException(ex);
-
-        StreamInput in = out.bytes().streamInput();
-        ParsingException e = in.readException();
-        assertEquals(ex.getIndex(), e.getIndex());
-        assertEquals(ex.getMessage(), e.getMessage());
-        assertEquals(ex.getLineNumber(), e.getLineNumber());
-        assertEquals(ex.getColumnNumber(), e.getColumnNumber());
-    }
-
-    public void testSerializeUnknownException() throws IOException {
-        BytesStreamOutput out = new BytesStreamOutput();
-        ParsingException parsingException = new ParsingException(1, 2, "foobar", null);
-        final Exception ex = new UnknownException("eggplant", parsingException);
-        out.writeException(ex);
-
-        StreamInput in = out.bytes().streamInput();
-        Throwable throwable = in.readException();
-        assertEquals("unknown_exception: eggplant", throwable.getMessage());
-        assertTrue(throwable instanceof ElasticsearchException);
-        ParsingException e = (ParsingException)throwable.getCause();
-                assertEquals(parsingException.getIndex(), e.getIndex());
-        assertEquals(parsingException.getMessage(), e.getMessage());
-        assertEquals(parsingException.getLineNumber(), e.getLineNumber());
-        assertEquals(parsingException.getColumnNumber(), e.getColumnNumber());
-    }
-
-    public void testWriteThrowable() throws IOException {
-
-        final QueryShardException queryShardException = new QueryShardException(new Index("foo", "_na_"), "foobar", null);
-        final UnknownException unknownException = new UnknownException("this exception is unknown", queryShardException);
-
-        final Exception[] causes = new Exception[]{
-                new IllegalStateException("foobar"),
-                new IllegalArgumentException("alalaal"),
-                new NullPointerException("boom"),
-                new EOFException("dadada"),
-                new ElasticsearchSecurityException("nono!"),
-                new NumberFormatException("not a number"),
-                new CorruptIndexException("baaaam booom", "this is my resource"),
-                new IndexFormatTooNewException("tooo new", 1, 2, 3),
-                new IndexFormatTooOldException("tooo new", 1, 2, 3),
-                new IndexFormatTooOldException("tooo new", "very old version"),
-                new ArrayIndexOutOfBoundsException("booom"),
-                new StringIndexOutOfBoundsException("booom"),
-                new FileNotFoundException("booom"),
-                new NoSuchFileException("booom"),
-                new AlreadyClosedException("closed!!", new NullPointerException()),
-                new LockObtainFailedException("can't lock directory", new NullPointerException()),
-                unknownException};
-        for (final Exception cause : causes) {
-            BytesStreamOutput out = new BytesStreamOutput();
-            ElasticsearchException ex = new ElasticsearchException("topLevel", cause);
-            out.writeException(ex);
-            StreamInput in = out.bytes().streamInput();
-            ElasticsearchException e = in.readException();
-            assertEquals(e.getMessage(), ex.getMessage());
-            assertTrue("Expected: " + e.getCause().getMessage() + " to contain: " +
-                            ex.getCause().getClass().getName() + " but it didn't",
-                    e.getCause().getMessage().contains(ex.getCause().getMessage()));
-            if (ex.getCause().getClass() != UnknownException.class) { // unknown exception is not directly mapped
-                assertEquals(e.getCause().getClass(), ex.getCause().getClass());
-            } else {
-                assertEquals(e.getCause().getClass(), NotSerializableExceptionWrapper.class);
-            }
-            assertArrayEquals(e.getStackTrace(), ex.getStackTrace());
-            assertTrue(e.getStackTrace().length > 1);
-            ElasticsearchAssertions.assertVersionSerializable(VersionUtils.randomVersion(random()), cause);
-            ElasticsearchAssertions.assertVersionSerializable(VersionUtils.randomVersion(random()), ex);
-            ElasticsearchAssertions.assertVersionSerializable(VersionUtils.randomVersion(random()), e);
-        }
-    }
-
-}
diff --git a/core/src/test/java/org/elasticsearch/ElasticsearchExceptionTests.java b/core/src/test/java/org/elasticsearch/ElasticsearchExceptionTests.java
index 576b53ad0e..4a8a90afab 100644
--- a/core/src/test/java/org/elasticsearch/ElasticsearchExceptionTests.java
+++ b/core/src/test/java/org/elasticsearch/ElasticsearchExceptionTests.java
@@ -21,73 +21,338 @@ package org.elasticsearch;
 
 import org.apache.lucene.util.Constants;
 import org.elasticsearch.action.RoutingMissingException;
+import org.elasticsearch.action.search.SearchPhaseExecutionException;
+import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationFailedException;
 import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.common.ParsingException;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentLocation;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.discovery.DiscoverySettings;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.query.QueryShardException;
 import org.elasticsearch.index.shard.IndexShardRecoveringException;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.SearchParseException;
+import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.TestSearchContext;
+import org.elasticsearch.transport.RemoteTransportException;
 import org.hamcrest.Matcher;
 
+import java.io.EOFException;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Collections;
 
 import static java.util.Collections.singleton;
-import static org.hamcrest.CoreMatchers.equalTo;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertToXContentEquivalent;
 import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.startsWith;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.startsWith;
 
 public class ElasticsearchExceptionTests extends ESTestCase {
 
+    public void testStatus() {
+        ElasticsearchException exception = new ElasticsearchException("test");
+        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
+
+        exception = new ElasticsearchException("test", new RuntimeException());
+        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
+
+        exception = new ElasticsearchException("test", new ResourceNotFoundException("test"));
+        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
+
+        exception = new RemoteTransportException("test", new ResourceNotFoundException("test"));
+        assertThat(exception.status(), equalTo(RestStatus.NOT_FOUND));
+
+        exception = new RemoteTransportException("test", new ResourceAlreadyExistsException("test"));
+        assertThat(exception.status(), equalTo(RestStatus.BAD_REQUEST));
+
+        exception = new RemoteTransportException("test", new IllegalArgumentException("foobar"));
+        assertThat(exception.status(), equalTo(RestStatus.BAD_REQUEST));
+
+        exception = new RemoteTransportException("test", new IllegalStateException("foobar"));
+        assertThat(exception.status(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
+    }
+
+    public void testGuessRootCause() {
+        {
+            ElasticsearchException exception = new ElasticsearchException("foo", new ElasticsearchException("bar",
+                    new IndexNotFoundException("foo", new RuntimeException("foobar"))));
+            ElasticsearchException[] rootCauses = exception.guessRootCauses();
+            assertEquals(rootCauses.length, 1);
+            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "index_not_found_exception");
+            assertEquals(rootCauses[0].getMessage(), "no such index");
+            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
+            ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
+            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed",
+                    new ShardSearchFailure[]{failure, failure1});
+            if (randomBoolean()) {
+                rootCauses = (randomBoolean() ? new RemoteTransportException("remoteboom", ex) : ex).guessRootCauses();
+            } else {
+                rootCauses = ElasticsearchException.guessRootCauses(randomBoolean() ? new RemoteTransportException("remoteboom", ex) : ex);
+            }
+            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "parsing_exception");
+            assertEquals(rootCauses[0].getMessage(), "foobar");
+
+            ElasticsearchException oneLevel = new ElasticsearchException("foo", new RuntimeException("foobar"));
+            rootCauses = oneLevel.guessRootCauses();
+            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "exception");
+            assertEquals(rootCauses[0].getMessage(), "foo");
+        }
+        {
+            ShardSearchFailure failure = new ShardSearchFailure(
+                    new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
+            ShardSearchFailure failure1 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 1));
+            ShardSearchFailure failure2 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 2));
+            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed",
+                    new ShardSearchFailure[]{failure, failure1, failure2});
+            final ElasticsearchException[] rootCauses = ex.guessRootCauses();
+            assertEquals(rootCauses.length, 2);
+            assertEquals(ElasticsearchException.getExceptionName(rootCauses[0]), "parsing_exception");
+            assertEquals(rootCauses[0].getMessage(), "foobar");
+            assertEquals(((ParsingException) rootCauses[0]).getLineNumber(), 1);
+            assertEquals(((ParsingException) rootCauses[0]).getColumnNumber(), 2);
+            assertEquals(ElasticsearchException.getExceptionName(rootCauses[1]), "query_shard_exception");
+            assertEquals((rootCauses[1]).getIndex().getName(), "foo1");
+            assertEquals(rootCauses[1].getMessage(), "foobar");
+        }
+
+        {
+            final ElasticsearchException[] foobars = ElasticsearchException.guessRootCauses(new IllegalArgumentException("foobar"));
+            assertEquals(foobars.length, 1);
+            assertTrue(foobars[0] instanceof ElasticsearchException);
+            assertEquals(foobars[0].getMessage(), "foobar");
+            assertEquals(foobars[0].getCause().getClass(), IllegalArgumentException.class);
+            assertEquals(foobars[0].getExceptionName(), "illegal_argument_exception");
+        }
+    }
+
+    public void testDeduplicate() throws IOException {
+        {
+            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
+            ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
+            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed",
+                    randomBoolean() ? failure1.getCause() : failure.getCause(), new ShardSearchFailure[]{failure, failure1});
+            XContentBuilder builder = XContentFactory.jsonBuilder();
+            builder.startObject();
+            ex.toXContent(builder, ToXContent.EMPTY_PARAMS);
+            builder.endObject();
+            String expected = "{\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\",\"phase\":\"search\"," +
+                    "\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\",\"reason\":" +
+                    "{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}}]}";
+            assertEquals(expected, builder.string());
+        }
+        {
+            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
+            ShardSearchFailure failure1 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 1));
+            ShardSearchFailure failure2 = new ShardSearchFailure(new QueryShardException(new Index("foo1", "_na_"), "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo1", "_na_"), 2));
+            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed",
+                    new ShardSearchFailure[]{failure, failure1, failure2});
+            XContentBuilder builder = XContentFactory.jsonBuilder();
+            builder.startObject();
+            ex.toXContent(builder, ToXContent.EMPTY_PARAMS);
+            builder.endObject();
+            String expected = "{\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\"," +
+                    "\"phase\":\"search\",\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\"," +
+                    "\"reason\":{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}},{\"shard\":1," +
+                    "\"index\":\"foo1\",\"node\":\"node_1\",\"reason\":{\"type\":\"query_shard_exception\",\"reason\":\"foobar\"," +
+                    "\"index_uuid\":\"_na_\",\"index\":\"foo1\"}}]}";
+            assertEquals(expected, builder.string());
+        }
+        {
+            ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
+            ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
+                    new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
+            NullPointerException nullPointerException = new NullPointerException();
+            SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed", nullPointerException,
+                    new ShardSearchFailure[]{failure, failure1});
+            assertEquals(nullPointerException, ex.getCause());
+            XContentBuilder builder = XContentFactory.jsonBuilder();
+            builder.startObject();
+            ex.toXContent(builder, ToXContent.EMPTY_PARAMS);
+            builder.endObject();
+            String expected = "{\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\"," +
+                    "\"phase\":\"search\",\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\"," +
+                    "\"reason\":{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}}]," +
+                    "\"caused_by\":{\"type\":\"null_pointer_exception\",\"reason\":null}}";
+            assertEquals(expected, builder.string());
+        }
+    }
+
+    /**
+     * Check whether this exception contains an exception of the given type:
+     * either it is of the given class itself or it contains a nested cause
+     * of the given type.
+     *
+     * @param exType the exception type to look for
+     * @return whether there is a nested exception of the specified type
+     */
+    private static boolean contains(Throwable t, Class<? extends Throwable> exType) {
+        if (exType == null) {
+            return false;
+        }
+        for (Throwable cause = t; t != null; t = t.getCause()) {
+            if (exType.isInstance(cause)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    public void testGetRootCause() {
+        Exception root = new RuntimeException("foobar");
+        ElasticsearchException exception = new ElasticsearchException("foo", new ElasticsearchException("bar",
+                new IllegalArgumentException("index is closed", root)));
+        assertEquals(root, exception.getRootCause());
+        assertTrue(contains(exception, RuntimeException.class));
+        assertFalse(contains(exception, EOFException.class));
+    }
+
+    public void testToString() {
+        ElasticsearchException exception = new ElasticsearchException("foo", new ElasticsearchException("bar",
+                new IllegalArgumentException("index is closed", new RuntimeException("foobar"))));
+        assertEquals("ElasticsearchException[foo]; nested: ElasticsearchException[bar]; nested: IllegalArgumentException" +
+                "[index is closed]; nested: RuntimeException[foobar];", exception.toString());
+    }
+
     public void testToXContent() throws IOException {
-        ElasticsearchException e = new ElasticsearchException("test");
-        assertExceptionAsJson(e, false, equalTo("{\"type\":\"exception\",\"reason\":\"test\"}"));
-
-        e = new IndexShardRecoveringException(new ShardId("_test", "_0", 5));
-        assertExceptionAsJson(e, false, equalTo("{\"type\":\"index_shard_recovering_exception\"," +
-                "\"reason\":\"CurrentState[RECOVERING] Already recovering\",\"index_uuid\":\"_0\",\"shard\":\"5\",\"index\":\"_test\"}"));
-
-        e = new BroadcastShardOperationFailedException(new ShardId("_index", "_uuid", 12), "foo", new IllegalStateException("bar"));
-        assertExceptionAsJson(e, false, equalTo("{\"type\":\"illegal_state_exception\",\"reason\":\"bar\"}"));
-
-        e = new ElasticsearchException(new IllegalArgumentException("foo"));
-        assertExceptionAsJson(e, false, equalTo("{\"type\":\"exception\",\"reason\":\"java.lang.IllegalArgumentException: foo\"," +
-                "\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"foo\"}}"));
-
-        e = new ElasticsearchException("foo", new IllegalStateException("bar"));
-        assertExceptionAsJson(e, false, equalTo("{\"type\":\"exception\",\"reason\":\"foo\"," +
-                "\"caused_by\":{\"type\":\"illegal_state_exception\",\"reason\":\"bar\"}}"));
-
-        // Test the same exception but with the "rest.exception.stacktrace.skip" parameter disabled: the stack_trace must be present
-        // in the JSON. Since the stack can be large, it only checks the beginning of the JSON.
-        assertExceptionAsJson(e, true, startsWith("{\"type\":\"exception\",\"reason\":\"foo\"," +
-                "\"caused_by\":{\"type\":\"illegal_state_exception\",\"reason\":\"bar\"," +
-                "\"stack_trace\":\"java.lang.IllegalStateException: bar" +
-                (Constants.WINDOWS ? "\\r\\n" : "\\n") +
-                "\\tat org.elasticsearch."));
+        {
+            ElasticsearchException e = new ElasticsearchException("test");
+            assertExceptionAsJson(e, "{\"type\":\"exception\",\"reason\":\"test\"}");
+        }
+        {
+            ElasticsearchException e = new IndexShardRecoveringException(new ShardId("_test", "_0", 5));
+            assertExceptionAsJson(e, "{\"type\":\"index_shard_recovering_exception\"," +
+                    "\"reason\":\"CurrentState[RECOVERING] Already recovering\",\"index_uuid\":\"_0\"," +
+                    "\"shard\":\"5\",\"index\":\"_test\"}");
+        }
+        {
+            ElasticsearchException e = new BroadcastShardOperationFailedException(new ShardId("_index", "_uuid", 12), "foo",
+                    new IllegalStateException("bar"));
+            assertExceptionAsJson(e, "{\"type\":\"illegal_state_exception\",\"reason\":\"bar\"}");
+        }
+        {
+            ElasticsearchException e = new ElasticsearchException(new IllegalArgumentException("foo"));
+            assertExceptionAsJson(e, "{\"type\":\"exception\",\"reason\":\"java.lang.IllegalArgumentException: foo\"," +
+                    "\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"foo\"}}");
+        }
+        {
+            ElasticsearchException e = new SearchParseException(new TestSearchContext(null), "foo", new XContentLocation(1,0));
+            assertExceptionAsJson(e, "{\"type\":\"search_parse_exception\",\"reason\":\"foo\",\"line\":1,\"col\":0}");
+        }
+        {
+            ElasticsearchException ex = new ElasticsearchException("foo",
+                    new ElasticsearchException("bar", new IllegalArgumentException("index is closed", new RuntimeException("foobar"))));
+            assertExceptionAsJson(ex, "{\"type\":\"exception\",\"reason\":\"foo\",\"caused_by\":{\"type\":\"exception\"," +
+                    "\"reason\":\"bar\",\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"index is closed\"," +
+                    "\"caused_by\":{\"type\":\"runtime_exception\",\"reason\":\"foobar\"}}}}");
+        }
+        {
+            ElasticsearchException e = new ElasticsearchException("foo", new IllegalStateException("bar"));
+            assertExceptionAsJson(e, "{\"type\":\"exception\",\"reason\":\"foo\"," +
+                    "\"caused_by\":{\"type\":\"illegal_state_exception\",\"reason\":\"bar\"}}");
+
+            // Test the same exception but with the "rest.exception.stacktrace.skip" parameter disabled: the stack_trace must be present
+            // in the JSON. Since the stack can be large, it only checks the beginning of the JSON.
+            ToXContent.Params params = new ToXContent.MapParams(
+                    Collections.singletonMap(ElasticsearchException.REST_EXCEPTION_SKIP_STACK_TRACE, "false"));
+            String actual;
+            try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
+                builder.startObject();
+                e.toXContent(builder, params);
+                builder.endObject();
+                actual = builder.string();
+            }
+            assertThat(actual, startsWith("{\"type\":\"exception\",\"reason\":\"foo\"," +
+                    "\"caused_by\":{\"type\":\"illegal_state_exception\",\"reason\":\"bar\"," +
+                    "\"stack_trace\":\"java.lang.IllegalStateException: bar" +
+                    (Constants.WINDOWS ? "\\r\\n" : "\\n") +
+                    "\\tat org.elasticsearch."));
+        }
     }
 
-    public void testToXContentWithHeaders() throws IOException {
+    public void testGenerateThrowableToXContent() throws IOException {
+        {
+            Exception ex;
+            if (randomBoolean()) {
+                // just a wrapper which is omitted
+                ex = new RemoteTransportException("foobar", new FileNotFoundException("foo not found"));
+            } else {
+                ex = new FileNotFoundException("foo not found");
+            }
+            assertExceptionAsJson(ex, "{\"type\":\"file_not_found_exception\",\"reason\":\"foo not found\"}");
+        }
+        {
+            ParsingException ex = new ParsingException(1, 2, "foobar", null);
+            assertExceptionAsJson(ex, "{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}");
+        }
+
+        { // test equivalence
+            ElasticsearchException ex =  new RemoteTransportException("foobar", new FileNotFoundException("foo not found"));
+            String toXContentString = Strings.toString(ex);
+            String throwableString = Strings.toString((builder, params) -> {
+                ElasticsearchException.generateThrowableXContent(builder, params, ex);
+                return builder;
+            });
+
+            assertEquals(throwableString, toXContentString);
+            assertEquals("{\"type\":\"file_not_found_exception\",\"reason\":\"foo not found\"}", toXContentString);
+        }
+
+        { // render header and metadata
+            ParsingException ex = new ParsingException(1, 2, "foobar", null);
+            ex.addMetadata("es.test1", "value1");
+            ex.addMetadata("es.test2", "value2");
+            ex.addHeader("test", "some value");
+            ex.addHeader("test_multi", "some value", "another value");
+            String expected = "{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2," +
+                    "\"test1\":\"value1\",\"test2\":\"value2\"," +
+                    "\"header\":{\"test_multi\":" +
+                    "[\"some value\",\"another value\"],\"test\":\"some value\"}}";
+            assertExceptionAsJson(ex, expected);
+        }
+    }
+
+    public void testToXContentWithHeadersAndMetadata() throws IOException {
         ElasticsearchException e = new ElasticsearchException("foo",
                                         new ElasticsearchException("bar",
                                                 new ElasticsearchException("baz",
                                                         new ClusterBlockException(singleton(DiscoverySettings.NO_MASTER_BLOCK_WRITES)))));
         e.addHeader("foo_0", "0");
         e.addHeader("foo_1", "1");
-        e.addHeader("es.header_foo_0", "foo_0");
-        e.addHeader("es.header_foo_1", "foo_1");
+        e.addMetadata("es.metadata_foo_0", "foo_0");
+        e.addMetadata("es.metadata_foo_1", "foo_1");
 
         final String expectedJson = "{"
             + "\"type\":\"exception\","
             + "\"reason\":\"foo\","
-            + "\"header_foo_0\":\"foo_0\","
-            + "\"header_foo_1\":\"foo_1\","
+            + "\"metadata_foo_0\":\"foo_0\","
+            + "\"metadata_foo_1\":\"foo_1\","
             + "\"caused_by\":{"
                 + "\"type\":\"exception\","
                 + "\"reason\":\"bar\","
@@ -106,7 +371,7 @@ public class ElasticsearchExceptionTests extends ESTestCase {
                 + "}"
         + "}";
 
-        assertExceptionAsJson(e, false, equalTo(expectedJson));
+        assertExceptionAsJson(e, expectedJson);
 
         ElasticsearchException parsed;
         try (XContentParser parser = createParser(XContentType.JSON.xContent(), expectedJson)) {
@@ -118,11 +383,12 @@ public class ElasticsearchExceptionTests extends ESTestCase {
 
         assertNotNull(parsed);
         assertEquals(parsed.getMessage(), "Elasticsearch exception [type=exception, reason=foo]");
-        assertThat(parsed.getHeaderKeys(), hasSize(4));
-        assertEquals(parsed.getHeader("header_foo_0").get(0), "foo_0");
-        assertEquals(parsed.getHeader("header_foo_1").get(0), "foo_1");
+        assertThat(parsed.getHeaderKeys(), hasSize(2));
         assertEquals(parsed.getHeader("foo_0").get(0), "0");
         assertEquals(parsed.getHeader("foo_1").get(0), "1");
+        assertThat(parsed.getMetadataKeys(), hasSize(2));
+        assertEquals(parsed.getMetadata("es.metadata_foo_0").get(0), "foo_0");
+        assertEquals(parsed.getMetadata("es.metadata_foo_1").get(0), "foo_1");
 
         ElasticsearchException cause = (ElasticsearchException) parsed.getCause();
         assertEquals(cause.getMessage(), "Elasticsearch exception [type=exception, reason=bar]");
@@ -185,24 +451,25 @@ public class ElasticsearchExceptionTests extends ESTestCase {
         cause = (ElasticsearchException) cause.getCause();
         assertEquals(cause.getMessage(),
                 "Elasticsearch exception [type=routing_missing_exception, reason=routing is required for [_test]/[_type]/[_id]]");
-        assertThat(cause.getHeaderKeys(), hasSize(2));
-        assertThat(cause.getHeader("index"), hasItem("_test"));
-        assertThat(cause.getHeader("index_uuid"), hasItem("_na_"));
+        assertThat(cause.getHeaderKeys(), hasSize(0));
+        assertThat(cause.getMetadataKeys(), hasSize(2));
+        assertThat(cause.getMetadata("es.index"), hasItem("_test"));
+        assertThat(cause.getMetadata("es.index_uuid"), hasItem("_na_"));
     }
 
-    public void testFromXContentWithHeaders() throws IOException {
+    public void testFromXContentWithHeadersAndMetadata() throws IOException {
         RoutingMissingException routing = new RoutingMissingException("_test", "_type", "_id");
         ElasticsearchException baz = new ElasticsearchException("baz", routing);
         baz.addHeader("baz_0", "baz0");
-        baz.addHeader("es.baz_1", "baz1");
+        baz.addMetadata("es.baz_1", "baz1");
         baz.addHeader("baz_2", "baz2");
-        baz.addHeader("es.baz_3", "baz3");
+        baz.addMetadata("es.baz_3", "baz3");
         ElasticsearchException bar = new ElasticsearchException("bar", baz);
-        bar.addHeader("es.bar_0", "bar0");
+        bar.addMetadata("es.bar_0", "bar0");
         bar.addHeader("bar_1", "bar1");
-        bar.addHeader("es.bar_2", "bar2");
+        bar.addMetadata("es.bar_2", "bar2");
         ElasticsearchException foo = new ElasticsearchException("foo", bar);
-        foo.addHeader("es.foo_0", "foo0");
+        foo.addMetadata("es.foo_0", "foo0");
         foo.addHeader("foo_1", "foo1");
 
         final XContent xContent = randomFrom(XContentType.values()).xContent();
@@ -218,31 +485,35 @@ public class ElasticsearchExceptionTests extends ESTestCase {
 
         assertNotNull(parsed);
         assertEquals(parsed.getMessage(), "Elasticsearch exception [type=exception, reason=foo]");
-        assertThat(parsed.getHeaderKeys(), hasSize(2));
-        assertThat(parsed.getHeader("foo_0"), hasItem("foo0"));
+        assertThat(parsed.getHeaderKeys(), hasSize(1));
         assertThat(parsed.getHeader("foo_1"), hasItem("foo1"));
+        assertThat(parsed.getMetadataKeys(), hasSize(1));
+        assertThat(parsed.getMetadata("es.foo_0"), hasItem("foo0"));
 
         ElasticsearchException cause = (ElasticsearchException) parsed.getCause();
         assertEquals(cause.getMessage(), "Elasticsearch exception [type=exception, reason=bar]");
-        assertThat(cause.getHeaderKeys(), hasSize(3));
-        assertThat(cause.getHeader("bar_0"), hasItem("bar0"));
+        assertThat(cause.getHeaderKeys(), hasSize(1));
         assertThat(cause.getHeader("bar_1"), hasItem("bar1"));
-        assertThat(cause.getHeader("bar_2"), hasItem("bar2"));
+        assertThat(cause.getMetadataKeys(), hasSize(2));
+        assertThat(cause.getMetadata("es.bar_0"), hasItem("bar0"));
+        assertThat(cause.getMetadata("es.bar_2"), hasItem("bar2"));
 
         cause = (ElasticsearchException) cause.getCause();
         assertEquals(cause.getMessage(), "Elasticsearch exception [type=exception, reason=baz]");
-        assertThat(cause.getHeaderKeys(), hasSize(4));
+        assertThat(cause.getHeaderKeys(), hasSize(2));
         assertThat(cause.getHeader("baz_0"), hasItem("baz0"));
-        assertThat(cause.getHeader("baz_1"), hasItem("baz1"));
         assertThat(cause.getHeader("baz_2"), hasItem("baz2"));
-        assertThat(cause.getHeader("baz_3"), hasItem("baz3"));
+        assertThat(cause.getMetadataKeys(), hasSize(2));
+        assertThat(cause.getMetadata("es.baz_1"), hasItem("baz1"));
+        assertThat(cause.getMetadata("es.baz_3"), hasItem("baz3"));
 
         cause = (ElasticsearchException) cause.getCause();
         assertEquals(cause.getMessage(),
                 "Elasticsearch exception [type=routing_missing_exception, reason=routing is required for [_test]/[_type]/[_id]]");
-        assertThat(cause.getHeaderKeys(), hasSize(2));
-        assertThat(cause.getHeader("index"), hasItem("_test"));
-        assertThat(cause.getHeader("index_uuid"), hasItem("_na_"));
+        assertThat(cause.getHeaderKeys(), hasSize(0));
+        assertThat(cause.getMetadataKeys(), hasSize(2));
+        assertThat(cause.getMetadata("es.index"), hasItem("_test"));
+        assertThat(cause.getMetadata("es.index_uuid"), hasItem("_na_"));
     }
 
     /**
@@ -251,17 +522,15 @@ public class ElasticsearchExceptionTests extends ESTestCase {
      * By default, the stack trace of the exception is not rendered. The parameter `errorTrace` forces the stack trace to
      * be rendered like the REST API does when the "error_trace" parameter is set to true.
      */
-    private static void assertExceptionAsJson(ElasticsearchException e, boolean errorTrace, Matcher<String> expected)
-            throws IOException {
-        ToXContent.Params params = ToXContent.EMPTY_PARAMS;
-        if (errorTrace) {
-            params = new ToXContent.MapParams(Collections.singletonMap(ElasticsearchException.REST_EXCEPTION_SKIP_STACK_TRACE, "false"));
-        }
-        try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
-            builder.startObject();
-            e.toXContent(builder, params);
-            builder.endObject();
-            assertThat(builder.bytes().utf8ToString(), expected);
-        }
+    private static void assertToXContentAsJson(ToXContent e, String expectedJson) throws IOException {
+        BytesReference actual = XContentHelper.toXContent(e, XContentType.JSON, randomBoolean());
+        assertToXContentEquivalent(new BytesArray(expectedJson), actual, XContentType.JSON);
+    }
+
+    private static void assertExceptionAsJson(Exception e, String expectedJson) throws IOException {
+        assertToXContentAsJson((builder, params) -> {
+            ElasticsearchException.generateThrowableXContent(builder, params, e);
+            return builder;
+        }, expectedJson);
     }
 }
diff --git a/core/src/test/java/org/elasticsearch/ExceptionSerializationTests.java b/core/src/test/java/org/elasticsearch/ExceptionSerializationTests.java
index f7660e3a2e..9dee89639a 100644
--- a/core/src/test/java/org/elasticsearch/ExceptionSerializationTests.java
+++ b/core/src/test/java/org/elasticsearch/ExceptionSerializationTests.java
@@ -18,6 +18,11 @@
  */
 package org.elasticsearch;
 
+import org.apache.lucene.index.CorruptIndexException;
+import org.apache.lucene.index.IndexFormatTooNewException;
+import org.apache.lucene.index.IndexFormatTooOldException;
+import org.apache.lucene.store.AlreadyClosedException;
+import org.apache.lucene.store.LockObtainFailedException;
 import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.RoutingMissingException;
 import org.elasticsearch.action.TimestampParsingException;
@@ -33,8 +38,11 @@ import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardRoutingState;
 import org.elasticsearch.cluster.routing.TestShardRouting;
 import org.elasticsearch.common.ParsingException;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.breaker.CircuitBreakingException;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.PathUtils;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.NotSerializableExceptionWrapper;
@@ -44,9 +52,6 @@ import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.CancellableThreadsTests;
 import org.elasticsearch.common.util.set.Sets;
-import org.elasticsearch.common.xcontent.ToXContent;
-import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentLocation;
 import org.elasticsearch.discovery.DiscoverySettings;
 import org.elasticsearch.env.ShardLockObtainFailedException;
@@ -81,6 +86,8 @@ import org.elasticsearch.transport.ActionTransportException;
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.TcpTransport;
 
+import java.io.EOFException;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.nio.file.AccessDeniedException;
@@ -97,6 +104,7 @@ import java.nio.file.NotDirectoryException;
 import java.nio.file.Path;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -107,6 +115,7 @@ import static java.lang.reflect.Modifier.isInterface;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.emptySet;
 import static java.util.Collections.singleton;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertVersionSerializable;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
 
@@ -511,30 +520,16 @@ public class ExceptionSerializationTests extends ESTestCase {
         assertEquals(1, ex.blocks().size());
     }
 
-    private String toXContent(ToXContent x) {
-        try {
-            XContentBuilder builder = XContentFactory.jsonBuilder();
-            builder.startObject();
-            x.toXContent(builder, ToXContent.EMPTY_PARAMS);
-            builder.endObject();
-            return builder.string();
-        } catch (IOException e) {
-            return "{ \"error\" : \"" + e.getMessage() + "\"}";
-        }
-    }
-
     public void testNotSerializableExceptionWrapper() throws IOException {
         NotSerializableExceptionWrapper ex = serialize(new NotSerializableExceptionWrapper(new NullPointerException()));
-        assertEquals("{\"type\":\"null_pointer_exception\",\"reason\":\"null_pointer_exception: null\"}", toXContent(ex));
+        assertEquals("{\"type\":\"null_pointer_exception\",\"reason\":\"null_pointer_exception: null\"}", Strings.toString(ex));
         ex = serialize(new NotSerializableExceptionWrapper(new IllegalArgumentException("nono!")));
-        assertEquals("{\"type\":\"illegal_argument_exception\",\"reason\":\"illegal_argument_exception: nono!\"}", toXContent(ex));
+        assertEquals("{\"type\":\"illegal_argument_exception\",\"reason\":\"illegal_argument_exception: nono!\"}", Strings.toString(ex));
 
         class UnknownException extends Exception {
-
-            public UnknownException(final String message) {
+            UnknownException(final String message) {
                 super(message);
             }
-
         }
 
         Exception[] unknowns = new Exception[]{
@@ -559,28 +554,94 @@ public class ExceptionSerializationTests extends ESTestCase {
         }
     }
 
+    public void testUnknownException() throws IOException {
+        ParsingException parsingException = new ParsingException(1, 2, "foobar", null);
+        final Exception ex = new UnknownException("eggplant", parsingException);
+        Exception exception = serialize(ex);
+        assertEquals("unknown_exception: eggplant", exception.getMessage());
+        assertTrue(exception instanceof ElasticsearchException);
+        ParsingException e = (ParsingException)exception.getCause();
+        assertEquals(parsingException.getIndex(), e.getIndex());
+        assertEquals(parsingException.getMessage(), e.getMessage());
+        assertEquals(parsingException.getLineNumber(), e.getLineNumber());
+        assertEquals(parsingException.getColumnNumber(), e.getColumnNumber());
+    }
+
+    public void testWriteThrowable() throws IOException {
+        final QueryShardException queryShardException = new QueryShardException(new Index("foo", "_na_"), "foobar", null);
+        final UnknownException unknownException = new UnknownException("this exception is unknown", queryShardException);
+
+        final Exception[] causes = new Exception[]{
+                new IllegalStateException("foobar"),
+                new IllegalArgumentException("alalaal"),
+                new NullPointerException("boom"),
+                new EOFException("dadada"),
+                new ElasticsearchSecurityException("nono!"),
+                new NumberFormatException("not a number"),
+                new CorruptIndexException("baaaam booom", "this is my resource"),
+                new IndexFormatTooNewException("tooo new", 1, 2, 3),
+                new IndexFormatTooOldException("tooo new", 1, 2, 3),
+                new IndexFormatTooOldException("tooo new", "very old version"),
+                new ArrayIndexOutOfBoundsException("booom"),
+                new StringIndexOutOfBoundsException("booom"),
+                new FileNotFoundException("booom"),
+                new NoSuchFileException("booom"),
+                new AlreadyClosedException("closed!!", new NullPointerException()),
+                new LockObtainFailedException("can't lock directory", new NullPointerException()),
+                unknownException};
+        for (final Exception cause : causes) {
+            ElasticsearchException ex = new ElasticsearchException("topLevel", cause);
+            ElasticsearchException deserialized = serialize(ex);
+            assertEquals(deserialized.getMessage(), ex.getMessage());
+            assertTrue("Expected: " + deserialized.getCause().getMessage() + " to contain: " +
+                            ex.getCause().getClass().getName() + " but it didn't",
+                    deserialized.getCause().getMessage().contains(ex.getCause().getMessage()));
+            if (ex.getCause().getClass() != UnknownException.class) { // unknown exception is not directly mapped
+                assertEquals(deserialized.getCause().getClass(), ex.getCause().getClass());
+            } else {
+                assertEquals(deserialized.getCause().getClass(), NotSerializableExceptionWrapper.class);
+            }
+            assertArrayEquals(deserialized.getStackTrace(), ex.getStackTrace());
+            assertTrue(deserialized.getStackTrace().length > 1);
+            assertVersionSerializable(VersionUtils.randomVersion(random()), cause);
+            assertVersionSerializable(VersionUtils.randomVersion(random()), ex);
+            assertVersionSerializable(VersionUtils.randomVersion(random()), deserialized);
+        }
+    }
+
     public void testWithRestHeadersException() throws IOException {
-        ElasticsearchException ex = new ElasticsearchException("msg");
-        ex.addHeader("foo", "foo", "bar");
-        ex = serialize(ex);
-        assertEquals("msg", ex.getMessage());
-        assertEquals(2, ex.getHeader("foo").size());
-        assertEquals("foo", ex.getHeader("foo").get(0));
-        assertEquals("bar", ex.getHeader("foo").get(1));
-
-        RestStatus status = randomFrom(RestStatus.values());
-        // ensure we are carrying over the headers even if not serialized
-        UnknownHeaderException uhe = new UnknownHeaderException("msg", status);
-        uhe.addHeader("foo", "foo", "bar");
-
-        ElasticsearchException serialize = serialize((ElasticsearchException) uhe);
-        assertTrue(serialize instanceof NotSerializableExceptionWrapper);
-        NotSerializableExceptionWrapper e = (NotSerializableExceptionWrapper) serialize;
-        assertEquals("unknown_header_exception: msg", e.getMessage());
-        assertEquals(2, e.getHeader("foo").size());
-        assertEquals("foo", e.getHeader("foo").get(0));
-        assertEquals("bar", e.getHeader("foo").get(1));
-        assertSame(status, e.status());
+        {
+            ElasticsearchException ex = new ElasticsearchException("msg");
+            ex.addHeader("foo", "foo", "bar");
+            ex.addMetadata("es.foo_metadata", "value1", "value2");
+            ex = serialize(ex);
+            assertEquals("msg", ex.getMessage());
+            assertEquals(2, ex.getHeader("foo").size());
+            assertEquals("foo", ex.getHeader("foo").get(0));
+            assertEquals("bar", ex.getHeader("foo").get(1));
+            assertEquals(2, ex.getMetadata("es.foo_metadata").size());
+            assertEquals("value1", ex.getMetadata("es.foo_metadata").get(0));
+            assertEquals("value2", ex.getMetadata("es.foo_metadata").get(1));
+        }
+        {
+            RestStatus status = randomFrom(RestStatus.values());
+            // ensure we are carrying over the headers and metadata even if not serialized
+            UnknownHeaderException uhe = new UnknownHeaderException("msg", status);
+            uhe.addHeader("foo", "foo", "bar");
+            uhe.addMetadata("es.foo_metadata", "value1", "value2");
+
+            ElasticsearchException serialize = serialize((ElasticsearchException) uhe);
+            assertTrue(serialize instanceof NotSerializableExceptionWrapper);
+            NotSerializableExceptionWrapper e = (NotSerializableExceptionWrapper) serialize;
+            assertEquals("unknown_header_exception: msg", e.getMessage());
+            assertEquals(2, e.getHeader("foo").size());
+            assertEquals("foo", e.getHeader("foo").get(0));
+            assertEquals("bar", e.getHeader("foo").get(1));
+            assertEquals(2, e.getMetadata("es.foo_metadata").size());
+            assertEquals("value1", e.getMetadata("es.foo_metadata").get(0));
+            assertEquals("value2", e.getMetadata("es.foo_metadata").get(1));
+            assertSame(status, e.status());
+        }
     }
 
     public void testNoLongerPrimaryShardException() throws IOException {
@@ -594,7 +655,7 @@ public class ExceptionSerializationTests extends ESTestCase {
     public static class UnknownHeaderException extends ElasticsearchException {
         private final RestStatus status;
 
-        public UnknownHeaderException(String msg, RestStatus status) {
+        UnknownHeaderException(String msg, RestStatus status) {
             super(msg);
             this.status = status;
         }
@@ -857,5 +918,75 @@ public class ExceptionSerializationTests extends ESTestCase {
         assertEquals("shard_lock_obtain_failed_exception: [foo][1]: boom", ex.getMessage());
     }
 
+    public void testBWCHeadersAndMetadata() throws IOException {
+        //this is a request serialized with headers only, no metadata as they were added in 5.3.0
+        BytesReference decoded = new BytesArray(Base64.getDecoder().decode
+                ("AQ10ZXN0ICBtZXNzYWdlACYtb3JnLmVsYXN0aWNzZWFyY2guRXhjZXB0aW9uU2VyaWFsaXphdGlvblRlc3RzASBFeGNlcHRpb25TZXJpYWxpemF0aW9uVG" +
+                        "VzdHMuamF2YQR0ZXN03wYkc3VuLnJlZmxlY3QuTmF0aXZlTWV0aG9kQWNjZXNzb3JJbXBsAR1OYXRpdmVNZXRob2RBY2Nlc3NvckltcGwuamF2Y" +
+                        "QdpbnZva2Uw/v///w8kc3VuLnJlZmxlY3QuTmF0aXZlTWV0aG9kQWNjZXNzb3JJbXBsAR1OYXRpdmVNZXRob2RBY2Nlc3NvckltcGwuamF2YQZp" +
+                        "bnZva2U+KHN1bi5yZWZsZWN0LkRlbGVnYXRpbmdNZXRob2RBY2Nlc3NvckltcGwBIURlbGVnYXRpbmdNZXRob2RBY2Nlc3NvckltcGwuamF2YQZ" +
+                        "pbnZva2UrGGphdmEubGFuZy5yZWZsZWN0Lk1ldGhvZAELTWV0aG9kLmphdmEGaW52b2tl8QMzY29tLmNhcnJvdHNlYXJjaC5yYW5kb21pemVkdG" +
+                        "VzdGluZy5SYW5kb21pemVkUnVubmVyARVSYW5kb21pemVkUnVubmVyLmphdmEGaW52b2tlsQ01Y29tLmNhcnJvdHNlYXJjaC5yYW5kb21pemVkd" +
+                        "GVzdGluZy5SYW5kb21pemVkUnVubmVyJDgBFVJhbmRvbWl6ZWRSdW5uZXIuamF2YQhldmFsdWF0ZYsHNWNvbS5jYXJyb3RzZWFyY2gucmFuZG9t" +
+                        "aXplZHRlc3RpbmcuUmFuZG9taXplZFJ1bm5lciQ5ARVSYW5kb21pemVkUnVubmVyLmphdmEIZXZhbHVhdGWvBzZjb20uY2Fycm90c2VhcmNoLnJ" +
+                        "hbmRvbWl6ZWR0ZXN0aW5nLlJhbmRvbWl6ZWRSdW5uZXIkMTABFVJhbmRvbWl6ZWRSdW5uZXIuamF2YQhldmFsdWF0Zb0HOWNvbS5jYXJyb3RzZW" +
+                        "FyY2gucmFuZG9taXplZHRlc3RpbmcucnVsZXMuU3RhdGVtZW50QWRhcHRlcgEVU3RhdGVtZW50QWRhcHRlci5qYXZhCGV2YWx1YXRlJDVvcmcuY" +
+                        "XBhY2hlLmx1Y2VuZS51dGlsLlRlc3RSdWxlU2V0dXBUZWFyZG93bkNoYWluZWQkMQEhVGVzdFJ1bGVTZXR1cFRlYXJkb3duQ2hhaW5lZC5qYXZh" +
+                        "CGV2YWx1YXRlMTBvcmcuYXBhY2hlLmx1Y2VuZS51dGlsLkFic3RyYWN0QmVmb3JlQWZ0ZXJSdWxlJDEBHEFic3RyYWN0QmVmb3JlQWZ0ZXJSdWx" +
+                        "lLmphdmEIZXZhbHVhdGUtMm9yZy5hcGFjaGUubHVjZW5lLnV0aWwuVGVzdFJ1bGVUaHJlYWRBbmRUZXN0TmFtZSQxAR5UZXN0UnVsZVRocmVhZE" +
+                        "FuZFRlc3ROYW1lLmphdmEIZXZhbHVhdGUwN29yZy5hcGFjaGUubHVjZW5lLnV0aWwuVGVzdFJ1bGVJZ25vcmVBZnRlck1heEZhaWx1cmVzJDEBI" +
+                        "1Rlc3RSdWxlSWdub3JlQWZ0ZXJNYXhGYWlsdXJlcy5qYXZhCGV2YWx1YXRlQCxvcmcuYXBhY2hlLmx1Y2VuZS51dGlsLlRlc3RSdWxlTWFya0Zh" +
+                        "aWx1cmUkMQEYVGVzdFJ1bGVNYXJrRmFpbHVyZS5qYXZhCGV2YWx1YXRlLzljb20uY2Fycm90c2VhcmNoLnJhbmRvbWl6ZWR0ZXN0aW5nLnJ1bGV" +
+                        "zLlN0YXRlbWVudEFkYXB0ZXIBFVN0YXRlbWVudEFkYXB0ZXIuamF2YQhldmFsdWF0ZSREY29tLmNhcnJvdHNlYXJjaC5yYW5kb21pemVkdGVzdG" +
+                        "luZy5UaHJlYWRMZWFrQ29udHJvbCRTdGF0ZW1lbnRSdW5uZXIBFlRocmVhZExlYWtDb250cm9sLmphdmEDcnVu7wI0Y29tLmNhcnJvdHNlYXJja" +
+                        "C5yYW5kb21pemVkdGVzdGluZy5UaHJlYWRMZWFrQ29udHJvbAEWVGhyZWFkTGVha0NvbnRyb2wuamF2YRJmb3JrVGltZW91dGluZ1Rhc2urBjZj" +
+                        "b20uY2Fycm90c2VhcmNoLnJhbmRvbWl6ZWR0ZXN0aW5nLlRocmVhZExlYWtDb250cm9sJDMBFlRocmVhZExlYWtDb250cm9sLmphdmEIZXZhbHV" +
+                        "hdGXOAzNjb20uY2Fycm90c2VhcmNoLnJhbmRvbWl6ZWR0ZXN0aW5nLlJhbmRvbWl6ZWRSdW5uZXIBFVJhbmRvbWl6ZWRSdW5uZXIuamF2YQ1ydW" +
+                        "5TaW5nbGVUZXN0lAc1Y29tLmNhcnJvdHNlYXJjaC5yYW5kb21pemVkdGVzdGluZy5SYW5kb21pemVkUnVubmVyJDUBFVJhbmRvbWl6ZWRSdW5uZ" +
+                        "XIuamF2YQhldmFsdWF0ZaIGNWNvbS5jYXJyb3RzZWFyY2gucmFuZG9taXplZHRlc3RpbmcuUmFuZG9taXplZFJ1bm5lciQ2ARVSYW5kb21pemVk" +
+                        "UnVubmVyLmphdmEIZXZhbHVhdGXUBjVjb20uY2Fycm90c2VhcmNoLnJhbmRvbWl6ZWR0ZXN0aW5nLlJhbmRvbWl6ZWRSdW5uZXIkNwEVUmFuZG9" +
+                        "taXplZFJ1bm5lci5qYXZhCGV2YWx1YXRl3wYwb3JnLmFwYWNoZS5sdWNlbmUudXRpbC5BYnN0cmFjdEJlZm9yZUFmdGVyUnVsZSQxARxBYnN0cm" +
+                        "FjdEJlZm9yZUFmdGVyUnVsZS5qYXZhCGV2YWx1YXRlLTljb20uY2Fycm90c2VhcmNoLnJhbmRvbWl6ZWR0ZXN0aW5nLnJ1bGVzLlN0YXRlbWVud" +
+                        "EFkYXB0ZXIBFVN0YXRlbWVudEFkYXB0ZXIuamF2YQhldmFsdWF0ZSQvb3JnLmFwYWNoZS5sdWNlbmUudXRpbC5UZXN0UnVsZVN0b3JlQ2xhc3NO" +
+                        "YW1lJDEBG1Rlc3RSdWxlU3RvcmVDbGFzc05hbWUuamF2YQhldmFsdWF0ZSlOY29tLmNhcnJvdHNlYXJjaC5yYW5kb21pemVkdGVzdGluZy5ydWx" +
+                        "lcy5Ob1NoYWRvd2luZ09yT3ZlcnJpZGVzT25NZXRob2RzUnVsZSQxAShOb1NoYWRvd2luZ09yT3ZlcnJpZGVzT25NZXRob2RzUnVsZS5qYXZhCG" +
+                        "V2YWx1YXRlKE5jb20uY2Fycm90c2VhcmNoLnJhbmRvbWl6ZWR0ZXN0aW5nLnJ1bGVzLk5vU2hhZG93aW5nT3JPdmVycmlkZXNPbk1ldGhvZHNSd" +
+                        "WxlJDEBKE5vU2hhZG93aW5nT3JPdmVycmlkZXNPbk1ldGhvZHNSdWxlLmphdmEIZXZhbHVhdGUoOWNvbS5jYXJyb3RzZWFyY2gucmFuZG9taXpl" +
+                        "ZHRlc3RpbmcucnVsZXMuU3RhdGVtZW50QWRhcHRlcgEVU3RhdGVtZW50QWRhcHRlci5qYXZhCGV2YWx1YXRlJDljb20uY2Fycm90c2VhcmNoLnJ" +
+                        "hbmRvbWl6ZWR0ZXN0aW5nLnJ1bGVzLlN0YXRlbWVudEFkYXB0ZXIBFVN0YXRlbWVudEFkYXB0ZXIuamF2YQhldmFsdWF0ZSQ5Y29tLmNhcnJvdH" +
+                        "NlYXJjaC5yYW5kb21pemVkdGVzdGluZy5ydWxlcy5TdGF0ZW1lbnRBZGFwdGVyARVTdGF0ZW1lbnRBZGFwdGVyLmphdmEIZXZhbHVhdGUkM29yZ" +
+                        "y5hcGFjaGUubHVjZW5lLnV0aWwuVGVzdFJ1bGVBc3NlcnRpb25zUmVxdWlyZWQkMQEfVGVzdFJ1bGVBc3NlcnRpb25zUmVxdWlyZWQuamF2YQhl" +
+                        "dmFsdWF0ZTUsb3JnLmFwYWNoZS5sdWNlbmUudXRpbC5UZXN0UnVsZU1hcmtGYWlsdXJlJDEBGFRlc3RSdWxlTWFya0ZhaWx1cmUuamF2YQhldmF" +
+                        "sdWF0ZS83b3JnLmFwYWNoZS5sdWNlbmUudXRpbC5UZXN0UnVsZUlnbm9yZUFmdGVyTWF4RmFpbHVyZXMkMQEjVGVzdFJ1bGVJZ25vcmVBZnRlck" +
+                        "1heEZhaWx1cmVzLmphdmEIZXZhbHVhdGVAMW9yZy5hcGFjaGUubHVjZW5lLnV0aWwuVGVzdFJ1bGVJZ25vcmVUZXN0U3VpdGVzJDEBHVRlc3RSd" +
+                        "WxlSWdub3JlVGVzdFN1aXRlcy5qYXZhCGV2YWx1YXRlNjljb20uY2Fycm90c2VhcmNoLnJhbmRvbWl6ZWR0ZXN0aW5nLnJ1bGVzLlN0YXRlbWVu" +
+                        "dEFkYXB0ZXIBFVN0YXRlbWVudEFkYXB0ZXIuamF2YQhldmFsdWF0ZSREY29tLmNhcnJvdHNlYXJjaC5yYW5kb21pemVkdGVzdGluZy5UaHJlYWR" +
+                        "MZWFrQ29udHJvbCRTdGF0ZW1lbnRSdW5uZXIBFlRocmVhZExlYWtDb250cm9sLmphdmEDcnVu7wIQamF2YS5sYW5nLlRocmVhZAELVGhyZWFkLm" +
+                        "phdmEDcnVu6QUABAdoZWFkZXIyAQZ2YWx1ZTIKZXMuaGVhZGVyMwEGdmFsdWUzB2hlYWRlcjEBBnZhbHVlMQplcy5oZWFkZXI0AQZ2YWx1ZTQAA" +
+                        "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" +
+                        "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" +
+                        "AAAAA"));
+
+        try (StreamInput in = decoded.streamInput()) {
+            //randomize the version across released and unreleased ones
+            Version version = randomFrom(Version.V_5_0_0, Version.V_5_0_1, Version.V_5_0_2,
+                    Version.V_5_0_3_UNRELEASED, Version.V_5_1_1_UNRELEASED, Version.V_5_1_2_UNRELEASED, Version.V_5_2_0_UNRELEASED);
+            in.setVersion(version);
+            ElasticsearchException exception = new ElasticsearchException(in);
+            assertEquals("test  message", exception.getMessage());
+            //the headers received as part of a single set get split based on their prefix
+            assertEquals(2, exception.getHeaderKeys().size());
+            assertEquals("value1", exception.getHeader("header1").get(0));
+            assertEquals("value2", exception.getHeader("header2").get(0));
+            assertEquals(2, exception.getMetadataKeys().size());
+            assertEquals("value3", exception.getMetadata("es.header3").get(0));
+            assertEquals("value4", exception.getMetadata("es.header4").get(0));
+        }
+    }
 
+    private static class UnknownException extends Exception {
+        UnknownException(final String message, final Exception cause) {
+            super(message, cause);
+        }
+    }
 }
diff --git a/core/src/test/java/org/elasticsearch/action/support/replication/ReplicationResponseTests.java b/core/src/test/java/org/elasticsearch/action/support/replication/ReplicationResponseTests.java
index 3ff113fc0f..87eecc218f 100644
--- a/core/src/test/java/org/elasticsearch/action/support/replication/ReplicationResponseTests.java
+++ b/core/src/test/java/org/elasticsearch/action/support/replication/ReplicationResponseTests.java
@@ -340,10 +340,16 @@ public class ReplicationResponseTests extends ESTestCase {
             ElasticsearchException ex = (ElasticsearchException) cause;
             for (String name : ex.getHeaderKeys()) {
                 assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
-                assertEquals(name.replaceFirst("es.", ""), parser.currentName());
+                assertEquals(name, parser.currentName());
                 assertEquals(XContentParser.Token.VALUE_STRING, parser.nextToken());
                 assertEquals(ex.getHeader(name).get(0), parser.text());
             }
+            for (String name : ex.getMetadataKeys()) {
+                assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
+                assertEquals(name.replaceFirst("es.", ""), parser.currentName());
+                assertEquals(XContentParser.Token.VALUE_STRING, parser.nextToken());
+                assertEquals(ex.getMetadata(name).get(0), parser.text());
+            }
             if (ex.getCause() != null) {
                 assertEquals(XContentParser.Token.FIELD_NAME, parser.nextToken());
                 assertEquals("caused_by", parser.currentName());
diff --git a/core/src/test/java/org/elasticsearch/rest/BytesRestResponseTests.java b/core/src/test/java/org/elasticsearch/rest/BytesRestResponseTests.java
index 6e3d0025eb..4546a0fa11 100644
--- a/core/src/test/java/org/elasticsearch/rest/BytesRestResponseTests.java
+++ b/core/src/test/java/org/elasticsearch/rest/BytesRestResponseTests.java
@@ -45,11 +45,9 @@ import static org.hamcrest.Matchers.notNullValue;
 public class BytesRestResponseTests extends ESTestCase {
 
     class UnknownException extends Exception {
-
-        public UnknownException(final String message, final Throwable cause) {
+        UnknownException(final String message, final Throwable cause) {
             super(message, cause);
         }
-
     }
 
     public void testWithHeaders() throws Exception {
@@ -57,6 +55,7 @@ public class BytesRestResponseTests extends ESTestCase {
         RestChannel channel = randomBoolean() ? new DetailedExceptionRestChannel(request) : new SimpleExceptionRestChannel(request);
 
         BytesRestResponse response = new BytesRestResponse(channel, new WithHeadersException());
+        assertEquals(2, response.getHeaders().size());
         assertThat(response.getHeaders().get("n1"), notNullValue());
         assertThat(response.getHeaders().get("n1"), contains("v11", "v12"));
         assertThat(response.getHeaders().get("n2"), notNullValue());
@@ -217,6 +216,7 @@ public class BytesRestResponseTests extends ESTestCase {
             super("");
             this.addHeader("n1", "v11", "v12");
             this.addHeader("n2", "v21", "v22");
+            this.addMetadata("es.test", "value1", "value2");
         }
     }
 
diff --git a/modules/lang-painless/src/main/java/org/elasticsearch/painless/Debug.java b/modules/lang-painless/src/main/java/org/elasticsearch/painless/Debug.java
index 65effdb8a4..29861000e1 100644
--- a/modules/lang-painless/src/main/java/org/elasticsearch/painless/Debug.java
+++ b/modules/lang-painless/src/main/java/org/elasticsearch/painless/Debug.java
@@ -59,11 +59,11 @@ public class Debug {
         /**
          * Headers to be added to the {@link ScriptException} for structured rendering.
          */
-        Map<String, List<String>> getHeaders() {
-            Map<String, List<String>> headers = new TreeMap<>();
-            headers.put("es.class", singletonList(objectToExplain == null ? "null" : objectToExplain.getClass().getName()));
-            headers.put("es.to_string", singletonList(Objects.toString(objectToExplain)));
-            return headers;
+        Map<String, List<String>> getMetadata() {
+            Map<String, List<String>> metadata = new TreeMap<>();
+            metadata.put("es.class", singletonList(objectToExplain == null ? "null" : objectToExplain.getClass().getName()));
+            metadata.put("es.to_string", singletonList(Objects.toString(objectToExplain)));
+            return metadata;
         }
     }
 }
diff --git a/modules/lang-painless/src/main/java/org/elasticsearch/painless/ScriptImpl.java b/modules/lang-painless/src/main/java/org/elasticsearch/painless/ScriptImpl.java
index 89c8984aaa..76ae31ce42 100644
--- a/modules/lang-painless/src/main/java/org/elasticsearch/painless/ScriptImpl.java
+++ b/modules/lang-painless/src/main/java/org/elasticsearch/painless/ScriptImpl.java
@@ -123,7 +123,7 @@ final class ScriptImpl implements ExecutableScript, LeafSearchScript {
             return executable.execute(variables, scorer, doc, aggregationValue);
         // Note that it is safe to catch any of the following errors since Painless is stateless.
         } catch (Debug.PainlessExplainError e) {
-            throw convertToScriptException(e, e.getHeaders());
+            throw convertToScriptException(e, e.getMetadata());
         } catch (PainlessError | BootstrapMethodError | OutOfMemoryError | StackOverflowError | Exception e) {
             throw convertToScriptException(e, emptyMap());
         }
@@ -135,7 +135,7 @@ final class ScriptImpl implements ExecutableScript, LeafSearchScript {
      * @param t The throwable to build an exception around.
      * @return The generated ScriptException.
      */
-    private ScriptException convertToScriptException(Throwable t, Map<String, List<String>> headers) {
+    private ScriptException convertToScriptException(Throwable t, Map<String, List<String>> metadata) {
         // create a script stack: this is just the script portion
         List<String> scriptStack = new ArrayList<>();
         for (StackTraceElement element : t.getStackTrace()) {
@@ -179,8 +179,8 @@ final class ScriptImpl implements ExecutableScript, LeafSearchScript {
             name = executable.getName();
         }
         ScriptException scriptException = new ScriptException("runtime error", t, scriptStack, name, PainlessScriptEngineService.NAME);
-        for (Map.Entry<String, List<String>> header : headers.entrySet()) {
-            scriptException.addHeader(header.getKey(), header.getValue());
+        for (Map.Entry<String, List<String>> entry : metadata.entrySet()) {
+            scriptException.addMetadata(entry.getKey(), entry.getValue());
         }
         return scriptException;
     }
diff --git a/modules/lang-painless/src/test/java/org/elasticsearch/painless/DebugTests.java b/modules/lang-painless/src/test/java/org/elasticsearch/painless/DebugTests.java
index 3b948bd259..1b2eb25c49 100644
--- a/modules/lang-painless/src/test/java/org/elasticsearch/painless/DebugTests.java
+++ b/modules/lang-painless/src/test/java/org/elasticsearch/painless/DebugTests.java
@@ -39,14 +39,14 @@ public class DebugTests extends ScriptTestCase {
         Debug.PainlessExplainError e = expectScriptThrows(Debug.PainlessExplainError.class, () -> exec(
                 "Debug.explain(params.a)", params, true));
         assertSame(dummy, e.getObjectToExplain());
-        assertThat(e.getHeaders(), hasEntry("es.class", singletonList("java.lang.Object")));
-        assertThat(e.getHeaders(), hasEntry("es.to_string", singletonList(dummy.toString())));
+        assertThat(e.getMetadata(), hasEntry("es.class", singletonList("java.lang.Object")));
+        assertThat(e.getMetadata(), hasEntry("es.to_string", singletonList(dummy.toString())));
 
         // Null should be ok
         e = expectScriptThrows(Debug.PainlessExplainError.class, () -> exec("Debug.explain(null)"));
         assertNull(e.getObjectToExplain());
-        assertThat(e.getHeaders(), hasEntry("es.class", singletonList("null")));
-        assertThat(e.getHeaders(), hasEntry("es.to_string", singletonList("null")));
+        assertThat(e.getMetadata(), hasEntry("es.class", singletonList("null")));
+        assertThat(e.getMetadata(), hasEntry("es.to_string", singletonList("null")));
 
         // You can't catch the explain exception
         e = expectScriptThrows(Debug.PainlessExplainError.class, () -> exec(
@@ -64,15 +64,15 @@ public class DebugTests extends ScriptTestCase {
     public void testPainlessExplainErrorSerialization() throws IOException {
         Map<String, Object> params = singletonMap("a", "jumped over the moon");
         ScriptException e = expectThrows(ScriptException.class, () -> exec("Debug.explain(params.a)", params, true));
-        assertEquals(singletonList("java.lang.String"), e.getHeader("es.class"));
-        assertEquals(singletonList("jumped over the moon"), e.getHeader("es.to_string"));
+        assertEquals(singletonList("java.lang.String"), e.getMetadata("es.class"));
+        assertEquals(singletonList("jumped over the moon"), e.getMetadata("es.to_string"));
 
         try (BytesStreamOutput out = new BytesStreamOutput()) {
             out.writeException(e);
             try (StreamInput in = out.bytes().streamInput()) {
                 ElasticsearchException read = (ScriptException) in.readException();
-                assertEquals(singletonList("java.lang.String"), read.getHeader("es.class"));
-                assertEquals(singletonList("jumped over the moon"), read.getHeader("es.to_string"));
+                assertEquals(singletonList("java.lang.String"), read.getMetadata("es.class"));
+                assertEquals(singletonList("jumped over the moon"), read.getMetadata("es.to_string"));
             }
         }
     }