Refactor Netty4Utils#maybeDie (#33021)

jasontedor · web-flow · commit 67bfb765ee8b · 2018-08-22T10:18:07.000-04:00
In our Netty layer we have had to take extra precautions against Netty
catching throwables which prevents them from reaching the uncaught
exception handler. This code has taken on additional uses in NIO layer
and now in the scheduler engine because there are other components in
stack traces that could catch throwables and suppress them from reaching
the uncaught exception handler. This commit is a simple cleanup of the
iterative evolution of this code to refactor all uses into a single
method in ExceptionsHelper.
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpChannel.java b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpChannel.java
@@ -21,11 +21,11 @@
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelPromise;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.concurrent.CompletableContext;
 import org.elasticsearch.http.HttpChannel;
 import org.elasticsearch.http.HttpResponse;
-import org.elasticsearch.transport.netty4.Netty4Utils;
 
 import java.net.InetSocketAddress;
 
@@ -42,7 +42,7 @@ public class Netty4HttpChannel implements HttpChannel {
             } else {
                 Throwable cause = f.cause();
                 if (cause instanceof Error) {
-                    Netty4Utils.maybeDie(cause);
+                    ExceptionsHelper.maybeDieOnAnotherThread(cause);
                     closeContext.completeExceptionally(new Exception(cause));
                 } else {
                     closeContext.completeExceptionally((Exception) cause);
@@ -59,7 +59,7 @@ public void sendResponse(HttpResponse response, ActionListener<Void> listener) {
                 listener.onResponse(null);
             } else {
                 final Throwable cause = f.cause();
-                Netty4Utils.maybeDie(cause);
+                ExceptionsHelper.maybeDieOnAnotherThread(cause);
                 if (cause instanceof Error) {
                     listener.onFailure(new Exception(cause));
                 } else {
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestHandler.java b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestHandler.java
@@ -27,7 +27,6 @@
 import io.netty.handler.codec.http.FullHttpRequest;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.http.HttpPipelinedRequest;
-import org.elasticsearch.transport.netty4.Netty4Utils;
 
 @ChannelHandler.Sharable
 class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelinedRequest<FullHttpRequest>> {
@@ -58,7 +57,7 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest<Full
             if (request.decoderResult().isFailure()) {
                 Throwable cause = request.decoderResult().cause();
                 if (cause instanceof Error) {
-                    ExceptionsHelper.dieOnError(cause);
+                    ExceptionsHelper.maybeDieOnAnotherThread(cause);
                     serverTransport.incomingRequestError(httpRequest, channel, new Exception(cause));
                 } else {
                     serverTransport.incomingRequestError(httpRequest, channel, (Exception) cause);
@@ -74,7 +73,7 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest<Full
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        Netty4Utils.maybeDie(cause);
+        ExceptionsHelper.maybeDieOnAnotherThread(cause);
         Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
         if (cause instanceof Error) {
             serverTransport.onException(channel, new Exception(cause));
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerChannel.java b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerChannel.java
@@ -20,10 +20,10 @@
 package org.elasticsearch.http.netty4;
 
 import io.netty.channel.Channel;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.concurrent.CompletableContext;
 import org.elasticsearch.http.HttpServerChannel;
-import org.elasticsearch.transport.netty4.Netty4Utils;
 
 import java.net.InetSocketAddress;
 
@@ -40,7 +40,7 @@ public class Netty4HttpServerChannel implements HttpServerChannel {
             } else {
                 Throwable cause = f.cause();
                 if (cause instanceof Error) {
-                    Netty4Utils.maybeDie(cause);
+                    ExceptionsHelper.maybeDieOnAnotherThread(cause);
                     closeContext.completeExceptionally(new Exception(cause));
                 } else {
                     closeContext.completeExceptionally((Exception) cause);
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java
@@ -41,6 +41,7 @@
 import io.netty.handler.timeout.ReadTimeoutException;
 import io.netty.handler.timeout.ReadTimeoutHandler;
 import io.netty.util.AttributeKey;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.network.CloseableChannel;
 import org.elasticsearch.common.network.NetworkService;
@@ -338,7 +339,7 @@ protected void initChannel(Channel ch) throws Exception {
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-            Netty4Utils.maybeDie(cause);
+            ExceptionsHelper.maybeDieOnAnotherThread(cause);
             super.exceptionCaught(ctx, cause);
         }
     }
@@ -354,7 +355,7 @@ private ServerChannelExceptionHandler(Netty4HttpServerTransport transport) {
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-            Netty4Utils.maybeDie(cause);
+            ExceptionsHelper.maybeDieOnAnotherThread(cause);
             Netty4HttpServerChannel httpServerChannel = ctx.channel().attr(HTTP_SERVER_CHANNEL_KEY).get();
             if (cause instanceof Error) {
                 transport.onServerException(httpServerChannel, new Exception(cause));
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4MessageChannelHandler.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4MessageChannelHandler.java
@@ -68,7 +68,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        Netty4Utils.maybeDie(cause);
+        ExceptionsHelper.maybeDieOnAnotherThread(cause);
         final Throwable unwrapped = ExceptionsHelper.unwrap(cause, ElasticsearchException.class);
         final Throwable newCause = unwrapped != null ? unwrapped : cause;
         Netty4TcpChannel tcpChannel = ctx.channel().attr(Netty4Transport.CHANNEL_KEY).get();
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpChannel.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpChannel.java
@@ -22,6 +22,7 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelPromise;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.concurrent.CompletableContext;
@@ -45,7 +46,7 @@ public class Netty4TcpChannel implements TcpChannel {
             } else {
                 Throwable cause = f.cause();
                 if (cause instanceof Error) {
-                    Netty4Utils.maybeDie(cause);
+                    ExceptionsHelper.maybeDieOnAnotherThread(cause);
                     closeContext.completeExceptionally(new Exception(cause));
                 } else {
                     closeContext.completeExceptionally((Exception) cause);
@@ -97,7 +98,7 @@ public void sendMessage(BytesReference reference, ActionListener<Void> listener)
                 listener.onResponse(null);
             } else {
                 final Throwable cause = f.cause();
-                Netty4Utils.maybeDie(cause);
+                ExceptionsHelper.maybeDieOnAnotherThread(cause);
                 if (cause instanceof Error) {
                     listener.onFailure(new Exception(cause));
                 } else {
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpServerChannel.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpServerChannel.java
@@ -20,6 +20,7 @@
 package org.elasticsearch.transport.netty4;
 
 import io.netty.channel.Channel;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.concurrent.CompletableContext;
 import org.elasticsearch.transport.TcpServerChannel;
@@ -41,7 +42,7 @@ public class Netty4TcpServerChannel implements TcpServerChannel {
             } else {
                 Throwable cause = f.cause();
                 if (cause instanceof Error) {
-                    Netty4Utils.maybeDie(cause);
+                    ExceptionsHelper.maybeDieOnAnotherThread(cause);
                     closeContext.completeExceptionally(new Exception(cause));
                 } else {
                     closeContext.completeExceptionally((Exception) cause);
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java
@@ -38,6 +38,7 @@
 import io.netty.util.concurrent.Future;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.SuppressForbidden;
@@ -228,7 +229,7 @@ protected Netty4TcpChannel initiateChannel(DiscoveryNode node, ActionListener<Vo
         ChannelFuture channelFuture = bootstrap.connect(address);
         Channel channel = channelFuture.channel();
         if (channel == null) {
-            Netty4Utils.maybeDie(channelFuture.cause());
+            ExceptionsHelper.maybeDieOnAnotherThread(channelFuture.cause());
             throw new IOException(channelFuture.cause());
         }
         addClosedExceptionLogger(channel);
@@ -242,7 +243,7 @@ protected Netty4TcpChannel initiateChannel(DiscoveryNode node, ActionListener<Vo
             } else {
                 Throwable cause = f.cause();
                 if (cause instanceof Error) {
-                    Netty4Utils.maybeDie(cause);
+                    ExceptionsHelper.maybeDieOnAnotherThread(cause);
                     listener.onFailure(new Exception(cause));
                 } else {
                     listener.onFailure((Exception) cause);
@@ -307,7 +308,7 @@ protected void initChannel(Channel ch) throws Exception {
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-            Netty4Utils.maybeDie(cause);
+            ExceptionsHelper.maybeDieOnAnotherThread(cause);
             super.exceptionCaught(ctx, cause);
         }
     }
@@ -333,7 +334,7 @@ protected void initChannel(Channel ch) throws Exception {
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-            Netty4Utils.maybeDie(cause);
+            ExceptionsHelper.maybeDieOnAnotherThread(cause);
             super.exceptionCaught(ctx, cause);
         }
     }
@@ -351,7 +352,7 @@ private class ServerChannelExceptionHandler extends ChannelHandlerAdapter {
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-            Netty4Utils.maybeDie(cause);
+            ExceptionsHelper.maybeDieOnAnotherThread(cause);
             Netty4TcpServerChannel serverChannel = ctx.channel().attr(SERVER_CHANNEL_KEY).get();
             if (cause instanceof Error) {
                 onServerException(serverChannel, new Exception(cause));
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Utils.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Utils.java
@@ -27,20 +27,16 @@
 import io.netty.util.NettyRuntime;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
-import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefIterator;
-import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.logging.ESLoggerFactory;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Locale;
-import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class Netty4Utils {
@@ -161,34 +157,4 @@ public static void closeChannels(final Collection<Channel> channels) throws IOEx
         }
     }
 
-    /**
-     * If the specified cause is an unrecoverable error, this method will rethrow the cause on a separate thread so that it can not be
-     * caught and bubbles up to the uncaught exception handler.
-     *
-     * @param cause the throwable to test
-     */
-    public static void maybeDie(final Throwable cause) {
-        final Logger logger = ESLoggerFactory.getLogger(Netty4Utils.class);
-        final Optional<Error> maybeError = ExceptionsHelper.maybeError(cause, logger);
-        if (maybeError.isPresent()) {
-            /*
-             * 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.
-             */
-            try {
-                // try to log the current stack trace
-                final String formatted = ExceptionsHelper.formatStackTrace(Thread.currentThread().getStackTrace());
-                logger.error("fatal error on the network layer\n{}", formatted);
-            } finally {
-                new Thread(
-                        () -> {
-                            throw maybeError.get();
-                        })
-                        .start();
-            }
-        }
-    }
-
 }
diff --git a/plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/HttpReadWriteHandler.java b/plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/HttpReadWriteHandler.java
@@ -139,7 +139,7 @@ private void handleRequest(Object msg) {
             if (request.decoderResult().isFailure()) {
                 Throwable cause = request.decoderResult().cause();
                 if (cause instanceof Error) {
-                    ExceptionsHelper.dieOnError(cause);
+                    ExceptionsHelper.maybeDieOnAnotherThread(cause);
                     transport.incomingRequestError(httpRequest, nioHttpChannel, new Exception(cause));
                 } else {
                     transport.incomingRequestError(httpRequest, nioHttpChannel, (Exception) cause);
diff --git a/plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NettyAdaptor.java b/plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NettyAdaptor.java
@@ -73,7 +73,7 @@ public void close() throws Exception {
         closeFuture.await();
         if (closeFuture.isSuccess() == false) {
             Throwable cause = closeFuture.cause();
-            ExceptionsHelper.dieOnError(cause);
+            ExceptionsHelper.maybeDieOnAnotherThread(cause);
             throw (Exception) cause;
         }
     }
@@ -84,7 +84,7 @@ public void addCloseListener(BiConsumer<Void, Exception> listener) {
                 listener.accept(null, null);
             } else {
                 final Throwable cause = f.cause();
-                ExceptionsHelper.dieOnError(cause);
+                ExceptionsHelper.maybeDieOnAnotherThread(cause);
                 assert cause instanceof Exception;
                 listener.accept(null, (Exception) cause);
             }
diff --git a/plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NettyListener.java b/plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NettyListener.java
@@ -223,7 +223,7 @@ public static NettyListener fromBiConsumer(BiConsumer<Void, Exception> biConsume
                     biConsumer.accept(null, null);
                 } else {
                     if (cause instanceof Error) {
-                        ExceptionsHelper.dieOnError(cause);
+                        ExceptionsHelper.maybeDieOnAnotherThread(cause);
                         biConsumer.accept(null, new Exception(cause));
                     } else {
                         biConsumer.accept(null, (Exception) cause);
diff --git a/qa/die-with-dignity/src/test/java/org/elasticsearch/qa/die_with_dignity/DieWithDignityIT.java b/qa/die-with-dignity/src/test/java/org/elasticsearch/qa/die_with_dignity/DieWithDignityIT.java
@@ -90,22 +90,22 @@ public void testDieWithDignity() throws Exception {
 
         final Iterator<String> it = lines.iterator();
 
-        boolean fatalErrorOnTheNetworkLayer = false;
+        boolean fatalError = false;
         boolean fatalErrorInThreadExiting = false;
 
-        while (it.hasNext() && (fatalErrorOnTheNetworkLayer == false || fatalErrorInThreadExiting == false)) {
+        while (it.hasNext() && (fatalError == false || fatalErrorInThreadExiting == false)) {
             final String line = it.next();
-            if (line.contains("fatal error on the network layer")) {
-                fatalErrorOnTheNetworkLayer = true;
-            } else if (line.matches(".*\\[ERROR\\]\\[o.e.b.ElasticsearchUncaughtExceptionHandler\\] \\[node-0\\]"
+            if (line.matches(".*\\[ERROR\\]\\[o\\.e\\.ExceptionsHelper\\s*\\] \\[node-0\\] fatal error")) {
+                fatalError = true;
+            } else if (line.matches(".*\\[ERROR\\]\\[o\\.e\\.b\\.ElasticsearchUncaughtExceptionHandler\\] \\[node-0\\]"
                     + " fatal error in thread \\[Thread-\\d+\\], exiting$")) {
                 fatalErrorInThreadExiting = true;
                 assertTrue(it.hasNext());
                 assertThat(it.next(), equalTo("java.lang.OutOfMemoryError: die with dignity"));
             }
         }
 
-        assertTrue(fatalErrorOnTheNetworkLayer);
+        assertTrue(fatalError);
         assertTrue(fatalErrorInThreadExiting);
     }
 
diff --git a/server/src/main/java/org/elasticsearch/ExceptionsHelper.java b/server/src/main/java/org/elasticsearch/ExceptionsHelper.java
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/scheduler/SchedulerEngine.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/scheduler/SchedulerEngine.java

Original file line number	Diff line number	Diff line change
`@@ -73,7 +73,7 @@ public void close() throws Exception {`
`73`	`73`	`closeFuture.await();`
`74`	`74`	`if (closeFuture.isSuccess() == false) {`
`75`	`75`	`Throwable cause = closeFuture.cause();`
`76`		`- ExceptionsHelper.dieOnError(cause);`
	`76`	`+ ExceptionsHelper.maybeDieOnAnotherThread(cause);`
`77`	`77`	`throw (Exception) cause;`
`78`	`78`	`}`
`79`	`79`	`}`
`@@ -84,7 +84,7 @@ public void addCloseListener(BiConsumer<Void, Exception> listener) {`
`84`	`84`	`listener.accept(null, null);`
`85`	`85`	`} else {`
`86`	`86`	`final Throwable cause = f.cause();`
`87`		`- ExceptionsHelper.dieOnError(cause);`
	`87`	`+ ExceptionsHelper.maybeDieOnAnotherThread(cause);`
`88`	`88`	`assert cause instanceof Exception;`
`89`	`89`	`listener.accept(null, (Exception) cause);`
`90`	`90`	`}`