Die with dignity on the network layer

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestHandler.java

@@ -27,6 +27,7 @@
 import io.netty.handler.codec.http.FullHttpRequest;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.http.netty4.pipelining.HttpPipelinedRequest;
+import org.elasticsearch.transport.netty4.Netty4Utils;
 
 @ChannelHandler.Sharable
 class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<Object> {
@@ -72,6 +73,7 @@ protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Except
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        Netty4Utils.maybeDie(cause);
         serverTransport.exceptionCaught(ctx, cause);
     }
 

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -53,9 +53,9 @@
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.BoundTransportAddress;
-import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.transport.NetworkExceptionHelper;
 import org.elasticsearch.common.transport.PortsRange;
+import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
@@ -578,6 +578,12 @@ protected void initChannel(Channel ch) throws Exception {
             ch.pipeline().addLast("handler", requestHandler);
         }
 
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            Netty4Utils.maybeDie(cause);
+            super.exceptionCaught(ctx, cause);
+        }
+
     }
 
 }

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4MessageChannelHandler.java

@@ -80,6 +80,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        Netty4Utils.maybeDie(cause);
         transport.exceptionCaught(ctx, cause);
     }
 

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java

@@ -508,6 +508,12 @@ protected void initChannel(Channel ch) throws Exception {
             ch.pipeline().addLast("dispatcher", new Netty4MessageChannelHandler(Netty4Transport.this, ".client"));
         }
 
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            Netty4Utils.maybeDie(cause);
+            super.exceptionCaught(ctx, cause);
+        }
+
     }
 
     protected class ServerChannelInitializer extends ChannelInitializer<Channel> {
@@ -526,6 +532,13 @@ protected void initChannel(Channel ch) throws Exception {
             ch.pipeline().addLast("size", new Netty4SizeHeaderFrameDecoder());
             ch.pipeline().addLast("dispatcher", new Netty4MessageChannelHandler(Netty4Transport.this, name));
         }
+
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            Netty4Utils.maybeDie(cause);
+            super.exceptionCaught(ctx, cause);
+        }
+
     }
 
 }

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Utils.java

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.elasticsearch.transport.netty4;
 
 import io.netty.buffer.ByteBuf;
@@ -28,9 +29,13 @@
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefIterator;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.logging.ESLoggerFactory;
 
 import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
@@ -63,8 +68,7 @@ public static ByteBuf toByteBuf(final BytesReference reference) {
             return ((ByteBufBytesReference) reference).toByteBuf();
         } else {
             final BytesRefIterator iterator = reference.iterator();
-            // usually we have one, two, or three components
-            // from the header, the message, and a buffer
+            // usually we have one, two, or three components from the header, the message, and a buffer
             final List<ByteBuf> buffers = new ArrayList<>(3);
             try {
                 BytesRef slice;
@@ -118,4 +122,31 @@ public static void closeChannels(final Collection<Channel> channels) throws IOEx
         }
     }
 
+    public static void maybeDie(final Throwable cause) throws IOException {
+        if (cause instanceof Error) {
+            /*
+             * Here be dragons. We want to rethrow this so that it bubbles up to the uncaught exception handler. Yet, Netty wraps too many
+             * invocations of user-code in try/catch blocks that swallow all throwables. This means that a rethrow here will not bubble up
+             * to where we want it to. So, we fork a thread and throw the exception from there where Netty can not get to it. We do not wrap
+             * the exception so as to not lose the original cause during exit, so we give the thread a name based on the previous stack
+             * frame so that at least we know where it came from (in case logging the current stack trace fails).
+             */
+            try (
+                final StringWriter sw = new StringWriter();
+                final PrintWriter pw = new PrintWriter(sw)) {
+                // try to log the current stack trace
+                Arrays.stream(Thread.currentThread().getStackTrace()).skip(1).map(e -> "\tat " + e).forEach(pw::println);
+                ESLoggerFactory.getLogger(Netty4Utils.class).error("fatal error on the network layer\n{}", sw.toString());
+            } finally {
+                final StackTraceElement previous = Thread.currentThread().getStackTrace()[2];
+                new Thread(
+                    () -> {
+                        throw (Error) cause;
+                    },
+                    previous.getClassName() + "#" + previous.getMethodName())
+                    .start();
+            }
+        }
+    }
+
 }

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpClient.java

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.elasticsearch.http.netty4;
 
 import io.netty.bootstrap.Bootstrap;