Add serialization support for InterruptedException

it's an important exception to serialize and we see it often in tests
etc. but then it's wrapped in NotSerializableExceptionWrapper which is
odd. This commit adds native support for this exception.

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

@@ -542,6 +542,8 @@ public abstract class StreamInput extends InputStream {
                     return (T) readStackTrace(new IllegalStateException(readOptionalString(), readThrowable()), this);
                 case 17:
                     return (T) readStackTrace(new LockObtainFailedException(readOptionalString(), readThrowable()), this);
+                case 18:
+                    return (T) readStackTrace(new InterruptedException(readOptionalString()), this);
                 default:
                     assert false : "no such exception for id: " + key;
             }

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

@@ -590,6 +590,9 @@ public abstract class StreamOutput extends OutputStream {
                 writeVInt(16);
             } else if (throwable instanceof LockObtainFailedException) {
                 writeVInt(17);
+            } else if (throwable instanceof InterruptedException) {
+                writeVInt(18);
+                writeCause = false;
             } else {
                 ElasticsearchException ex;
                 final String name = throwable.getClass().getName();

core/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java

@@ -497,7 +497,6 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
         serverBootstrap.setOption("child.receiveBufferSizePredictorFactory", receiveBufferSizePredictorFactory);
         serverBootstrap.setOption("reuseAddress", reuseAddress);
         serverBootstrap.setOption("child.reuseAddress", reuseAddress);
-
         serverBootstraps.put(name, serverBootstrap);
     }
 

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

@@ -607,4 +607,20 @@ public class ExceptionSerializationTests extends ESTestCase {
         assertEquals(ex.status(), e.status());
         assertEquals(RestStatus.UNAUTHORIZED, e.status());
     }
+
+    public void testInterruptedException() throws IOException {
+        InterruptedException orig = randomBoolean() ? new InterruptedException("boom") : new InterruptedException();
+        InterruptedException ex = serialize(orig);
+        assertEquals(orig.getMessage(), ex.getMessage());
+    }
+
+    public static class UnknownException extends Exception {
+        public UnknownException(String message) {
+            super(message);
+        }
+
+        public UnknownException(String message, Throwable cause) {
+            super(message, cause);
+        }
+    }
 }